build: bump version to 0.8.1

Cargo.lock

@@ -1532,7 +1532,7 @@ checksum = "38bf9645c8b145698bb0b18a4637dcacbc421ea49bef2317e4fd8065a387cf21"
 
 [[package]]
 name = "nostr-rs-relay"
-version = "0.8.0"
+version = "0.8.1"
 dependencies = [
  "anyhow",
  "async-std",

Cargo.toml

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

config.toml

@@ -81,7 +81,8 @@ reject_future_seconds = 1800
 
 # Limit client subscriptions created per second, averaged over one
 # minute.  Must be an integer.  If not set (or set to 0), defaults to
-# unlimited.
+# unlimited.  Strongly recommended to set this to a low value such as
+# 10 to ensure fair service.
 #subscriptions_per_min = 0
 
 # UNIMPLEMENTED...

src/info.rs

@@ -35,7 +35,7 @@ impl From<config::Info> for RelayInfo {
             description: i.description,
             pubkey: i.pubkey,
             contact: i.contact,
-            supported_nips: Some(vec![1, 2, 9, 11, 12, 15, 16, 20, 22]),
+            supported_nips: Some(vec![1, 2, 9, 11, 12, 15, 16, 20, 22, 33]),
             software: Some("https://git.sr.ht/~gheartsfield/nostr-rs-relay".to_owned()),
             version: CARGO_PKG_VERSION.map(std::borrow::ToOwned::to_owned),
         }

src/repo/sqlite.rs

@@ -344,27 +344,37 @@ impl NostrRepo for SqliteRepo {
                     db_queue_time, client_id, sub.id
                 );
             }
+            // check before getting a DB connection if the client still wants the results
+            if abandon_query_rx.try_recv().is_ok() {
+                debug!("query cancelled by client (before execution) (cid: {}, sub: {:?})", client_id, sub.id);
+                return Ok(());
+            }
+
             let start = Instant::now();
             let mut row_count: usize = 0;
             // cutoff for displaying slow queries
             let slow_cutoff = Duration::from_millis(250);
             let mut filter_count = 0;
             // remove duplicates from the filter list.
-            for filter in sub.filters.iter() {
-                let filter_start = Instant::now();
-                filter_count += 1;
-                let (q, p, idx) = query_from_filter(&filter);
-                let sql_gen_elapsed = start.elapsed();
-
-                if sql_gen_elapsed > Duration::from_millis(10) {
-                    debug!("SQL (slow) generated in {:?}", filter_start.elapsed());
+            if let Ok(mut conn) = self.read_pool.get() {
+                {
+                    let pool_state = self.read_pool.state();
+                    metrics.db_connections.set((pool_state.connections - pool_state.idle_connections).into());
                 }
-                // any client that doesn't cause us to generate new rows in 5
-                // seconds gets dropped.
-                let abort_cutoff = Duration::from_secs(5);
-                let mut slow_first_event;
-                let mut last_successful_send = Instant::now();
-                if let Ok(mut conn) = self.read_pool.get() {
+                for filter in sub.filters.iter() {
+                    let filter_start = Instant::now();
+                    filter_count += 1;
+                    let (q, p, idx) = query_from_filter(&filter);
+                    let sql_gen_elapsed = start.elapsed();
+
+                    if sql_gen_elapsed > Duration::from_millis(10) {
+                        debug!("SQL (slow) generated in {:?}", filter_start.elapsed());
+                    }
+                    // any client that doesn't cause us to generate new rows in 2
+                    // seconds gets dropped.
+                    let abort_cutoff = Duration::from_secs(2);
+                    let mut slow_first_event;
+                    let mut last_successful_send = Instant::now();
                     // execute the query.
                     // make the actual SQL query (with parameters inserted) available
                     conn.trace(Some(|x| {trace!("SQL trace: {:?}", x)}));
@@ -384,8 +394,8 @@ impl NostrRepo for SqliteRepo {
                             // to reduce logging; only show 1/16th of clients (leading 0)
                             if slow_first_event && client_id.starts_with('0') {
                                 debug!(
-                                    "filter first result (slow): {} (cid: {}, sub: {:?})",
-                                    serde_json::to_string(&filter)?, client_id, sub.id
+                                    "filter first result in {:?} (slow): {} (cid: {}, sub: {:?})",
+                                    first_event_elapsed, serde_json::to_string(&filter)?, client_id, sub.id
                                 );
                             }
                             first_result = false;
@@ -432,7 +442,8 @@ impl NostrRepo for SqliteRepo {
                                 return Ok(());
                             }
                             // give the queue a chance to clear before trying again
-                            thread::sleep(Duration::from_millis(100));
+                            debug!("query thread sleeping due to full query_tx (cid: {}, sub: {:?})", client_id, sub.id);
+                            thread::sleep(Duration::from_millis(500));
                         }
                         // TODO: we could use try_send, but we'd have to juggle
                         // getting the query result back as part of the error
@@ -445,17 +456,20 @@ impl NostrRepo for SqliteRepo {
                             .ok();
                         last_successful_send = Instant::now();
                     }
-                } else {
-                    warn!("Could not get a database connection for querying");
-                }
-                // if the filter took too much db_time, print out the JSON.
-                if filter_start.elapsed() > slow_cutoff && client_id.starts_with('0') {
-                    debug!(
-                        "query filter req (slow): {} (cid: {}, sub: {:?}, filter: {})",
-                        serde_json::to_string(&filter)?, client_id, sub.id, filter_count
-                    );
-                }
+                    metrics
+                        .query_db
+                        .observe(filter_start.elapsed().as_secs_f64());
+                    // if the filter took too much db_time, print out the JSON.
+                    if filter_start.elapsed() > slow_cutoff && client_id.starts_with('0') {
+                        debug!(
+                            "query filter req (slow): {} (cid: {}, sub: {:?}, filter: {})",
+                            serde_json::to_string(&filter)?, client_id, sub.id, filter_count
+                        );
+                    }
 
+                }
+            } else {
+                warn!("Could not get a database connection for querying");
             }
             drop(sem); // new query can begin
             debug!(

src/server.rs

@@ -14,6 +14,7 @@ use crate::nip05;
 use crate::notice::Notice;
 use crate::subscription::Subscription;
 use prometheus::IntCounterVec;
+use prometheus::IntGauge;
 use prometheus::{Encoder, Histogram, IntCounter, HistogramOpts, Opts, Registry, TextEncoder};
 use futures::SinkExt;
 use futures::StreamExt;
@@ -232,6 +233,10 @@ fn create_metrics() -> (Registry, NostrMetrics) {
         "nostr_query_seconds",
         "Subscription response times",
     )).unwrap();
+    let query_db = Histogram::with_opts(HistogramOpts::new(
+        "nostr_filter_seconds",
+        "Filter SQL query times",
+    )).unwrap();
     let write_events = Histogram::with_opts(HistogramOpts::new(
         "nostr_events_write_seconds",
         "Event writing response times",
@@ -244,6 +249,9 @@ fn create_metrics() -> (Registry, NostrMetrics) {
         "nostr_connections_total",
         "New connections",
     )).unwrap();
+    let db_connections = IntGauge::with_opts(Opts::new(
+        "nostr_db_connections", "Active database connections"
+    )).unwrap();
     let query_aborts = IntCounterVec::new(
         Opts::new("nostr_query_abort_total", "Aborted queries"),
         vec!["reason"].as_slice(),
@@ -265,9 +273,11 @@ fn create_metrics() -> (Registry, NostrMetrics) {
         vec!["reason"].as_slice(),
         ).unwrap();
     registry.register(Box::new(query_sub.clone())).unwrap();
+    registry.register(Box::new(query_db.clone())).unwrap();
     registry.register(Box::new(write_events.clone())).unwrap();
     registry.register(Box::new(sent_events.clone())).unwrap();
     registry.register(Box::new(connections.clone())).unwrap();
+    registry.register(Box::new(db_connections.clone())).unwrap();
     registry.register(Box::new(query_aborts.clone())).unwrap();
     registry.register(Box::new(cmd_req.clone())).unwrap();
     registry.register(Box::new(cmd_event.clone())).unwrap();
@@ -275,9 +285,11 @@ fn create_metrics() -> (Registry, NostrMetrics) {
     registry.register(Box::new(disconnects.clone())).unwrap();
     let metrics = NostrMetrics {
         query_sub,
+        query_db,
         write_events,
         sent_events,
         connections,
+        db_connections,
 	disconnects,
         query_aborts,
 	cmd_req,
@@ -564,7 +576,7 @@ async fn nostr_server(
     // we will send out the tx handle to any query we generate.
     // 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);
+    let (query_tx, mut query_rx) = mpsc::channel::<db::QueryResult>(20_000);
     // Create channel for receiving NOTICEs
     let (notice_tx, mut notice_rx) = mpsc::channel::<Notice>(128);
 
@@ -799,11 +811,11 @@ async fn nostr_server(
                         break;
                     }
                     Err(Error::EventMaxLengthError(s)) => {
-                        info!("client sent event larger ({} bytes) than max size (cid: {})", s, cid);
+                        info!("client sent command larger ({} bytes) than max size (cid: {})", s, cid);
                         ws_stream.send(make_notice_message(&Notice::message("event exceeded max size".into()))).await.ok();
                     },
                     Err(Error::ProtoParseError) => {
-                        info!("client sent event that could not be parsed (cid: {})", cid);
+                        info!("client sent command that could not be parsed (cid: {})", cid);
                         ws_stream.send(make_notice_message(&Notice::message("could not parse command".into()))).await.ok();
                     },
                     Err(e) => {
@@ -830,6 +842,8 @@ async fn nostr_server(
 #[derive(Clone)]
 pub struct NostrMetrics {
     pub query_sub: Histogram, // response time of successful subscriptions
+    pub query_db: Histogram, // individual database query execution time
+    pub db_connections: IntGauge, // database connections in use
     pub write_events: Histogram, // response time of event writes
     pub sent_events: IntCounterVec, // count of events sent to clients
     pub connections: IntCounter, // count of websocket connections