build: bump version to 0.7.15

Cargo.lock

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

Cargo.toml

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

docs/database-maintenance.md

@@ -0,0 +1,125 @@
+# Database Maintenance
+
+`nostr-rs-relay` uses the SQLite embedded database to minimize
+dependencies and overall footprint of running a relay.  If traffic is
+light, the relay should just run with very little need for
+intervention.  For heavily trafficked relays, there are a number of
+steps that the operator may need to take to maintain performance and
+limit disk usage.
+
+This maintenance guide is current as of version `0.7.14`.  Future
+versions may incorporate and automate some of these steps.
+
+## Backing Up the Database
+
+To prevent data loss, the database should be backed up regularly.  The
+recommended method is to use the `sqlite3` command to perform an
+"Online Backup".  This can be done while the relay is running, queries
+can still run and events will be persisted during the backup.
+
+The following commands will perform a backup of the database to a
+dated file, and then compress to minimize size:
+
+```console
+BACKUP_FILE=/var/backups/nostr/`date +%Y%m%d_%H%M`.db
+sqlite3 -readonly /apps/nostr-relay/nostr.db ".backup $BACKUP_FILE
+sqlite3 $BACKUP_FILE "vacuum;"
+bzip2 -9 $BACKUP_FILE
+```
+
+Nostr events are very compressible.  Expect a compression ratio on the
+order of 4:1, resulting in a 75% space saving.
+
+## Vacuuming the Database
+
+As the database is updated, it can become fragmented.  Performing a
+full `vacuum` will rebuild the entire database file, and can reduce
+space.  Running this may reduce the size of the database file,
+especially if a large amount of data was updated or deleted.
+
+```console
+vacuum;
+```
+
+## Clearing Hidden Events
+
+When events are deleted, either through deletion events, metadata or
+follower updates, or a replaceable event kind, the event is not
+actually removed from the database.  Instead, a flag `HIDDEN` is set
+to true for the event, which excludes it from search results.  The
+original intent was to ensure that subsequent rebroadcasts of the
+event would be easily detected as having been deleted, and would not
+need to be stored again.  In practice, this decision causes excessive
+growth of the `tags` table, since all the previous followers are
+retained for those `HIDDEN` events.
+
+The `event` and especially the `tag` table can be significantly
+reduced in size by running these commands:
+
+```console
+PRAGMA foreign_keys = ON;
+delete from event where HIDDEN=true;
+```
+
+## Manually Removing Events
+
+For a variety of reasons, an operator may wish to remove some events
+from the database.  The only way of achieving this today is with
+manually run SQL commands.
+
+It is recommended to have a good backup prior to manually running SQL
+commands!
+
+In all cases, it is mandatory to enable foreign keys, and this must be
+done for every connection.  Otherwise, you will likely orphan rows in
+the `tag` table.
+
+### Deleting Specific Event
+
+```console
+PRAGMA foreign_keys = ON;
+delete from event where event_hash=x'00000000000c1271675dc86e3e1dd1336827bccabb90dc4c9d3b4465efefe00e';
+```
+
+### Deleting All Events for Pubkey
+
+```console
+PRAGMA foreign_keys = ON;
+delete from event where author=x'000000000002c7831d9c5a99f183afc2813a6f69a16edda7f6fc0ed8110566e6';
+```
+
+### Deleting All Events of a Kind
+
+
+```console
+PRAGMA foreign_keys = ON;
+delete from event where kind=70202;
+```
+
+### Deleting Old Events
+
+In this scenario, we wish to delete any event that has been stored by
+our relay for more than 1 month.  Crucially, this is based on when the
+event was stored, not when the event says it was created.  If an event
+has a `created` field of 2 years ago, but was first sent to our relay
+yesterday, it would not be deleted in this scenario.  Keep in mind, we
+do not track anything for re-broadcast events that we already have, so
+this is not a very effective way of implementing a "least recently
+seen" policy.
+
+```console
+PRAGMA foreign_keys = ON;
+TODO!
+```
+
+### Delete Profile Events with No Recent Events
+
+Many users create profiles, post a "hello world" event, and then never
+appear again (likely using an ephemeral keypair that was lost in the
+browser cache).  We can find these accounts and remove them after some
+time.
+
+```console
+PRAGMA foreign_keys = ON;
+TODO!
+```

src/db.rs

@@ -19,8 +19,10 @@ use r2d2_sqlite::SqliteConnectionManager;
 use rusqlite::params;
 use rusqlite::types::ToSql;
 use rusqlite::OpenFlags;
+use tokio::sync::{Mutex, MutexGuard};
 use std::fmt::Write as _;
 use std::path::Path;
+use std::sync::Arc;
 use std::thread;
 use std::time::Duration;
 use std::time::Instant;
@@ -691,7 +693,7 @@ fn log_pool_stats(name: &str, pool: &SqlitePool) {
 
 
 /// Perform database maintenance on a regular basis
-pub async fn db_optimize(pool: SqlitePool) {
+pub async fn db_optimize_task(pool: SqlitePool) {
     tokio::task::spawn(async move {
         loop {
             tokio::select! {
@@ -710,7 +712,7 @@ pub async fn db_optimize(pool: SqlitePool) {
 }
 
 /// Perform database WAL checkpoint on a regular basis
-pub async fn db_checkpoint(pool: SqlitePool) {
+pub async fn db_checkpoint_task(pool: SqlitePool, safe_to_read: Arc<Mutex<u64>>) {
     tokio::task::spawn(async move {
 	// WAL size in pages.
 	let mut current_wal_size = 0;
@@ -719,11 +721,12 @@ pub async fn db_checkpoint(pool: SqlitePool) {
 	// default threshold for the busy timer
 	let busy_wait_default = Duration::from_secs(1);
 	// if the WAL file is getting too big, switch to this
-	let busy_wait_default_long = Duration::from_secs(5);
+	let busy_wait_default_long = Duration::from_secs(10);
         loop {
             tokio::select! {
                 _ = tokio::time::sleep(Duration::from_secs(CHECKPOINT_FREQ_SEC)) => {
                     if let Ok(mut conn) = pool.get() {
+                        let mut _guard:Option<MutexGuard<u64>> = None;
 			// the busy timer will block writers, so don't set
 			// this any higher than you want max latency for event
 			// writes.
@@ -732,6 +735,9 @@ pub async fn db_checkpoint(pool: SqlitePool) {
 			} else {
 			    // if the wal size has exceeded a threshold, increase the busy timeout.
 			    conn.busy_timeout(busy_wait_default_long).ok();
+			    // take a lock that will prevent new readers.
+			    info!("blocking new readers to perform wal_checkpoint");
+			    _guard = Some(safe_to_read.lock().await);
 			}
                         debug!("running wal_checkpoint(TRUNCATE)");
                         if let Ok(new_size) = checkpoint_db(&mut conn) {
@@ -756,9 +762,14 @@ pub async fn db_query(
     pool: SqlitePool,
     query_tx: tokio::sync::mpsc::Sender<QueryResult>,
     mut abandon_query_rx: tokio::sync::oneshot::Receiver<()>,
+    safe_to_read: Arc<Mutex<u64>>,
 ) {
     let pre_spawn_start = Instant::now();
     task::spawn_blocking(move || {
+	{
+	    // if we are waiting on a checkpoint, stop until it is complete
+	    let _ = safe_to_read.blocking_lock();
+	}
         let db_queue_time = pre_spawn_start.elapsed();
         // if the queue time was very long (>5 seconds), spare the DB and abort.
         if db_queue_time > Duration::from_secs(5) {
@@ -817,6 +828,17 @@ pub async fn db_query(
                         sub, client_id, sub.id
                     );
 		}
+		// check if a checkpoint is trying to run, and abort
+		if row_count % 100 == 0 {
+		    {
+			if let Err(_) = safe_to_read.try_lock() {
+			    // lock was held, abort this query
+			    debug!("query aborted due to checkpoint (cid: {}, sub: {:?})", client_id, sub.id);
+			    return Ok(());
+			}
+		    }
+		}
+
                 // 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);
@@ -838,6 +860,12 @@ pub async fn db_query(
                             let ok: Result<()> = Ok(());
                             return ok;
                         }
+			// check if a checkpoint is trying to run, and abort
+			if let Err(_) = safe_to_read.try_lock() {
+			    // lock was held, abort this query
+			    debug!("query aborted due to checkpoint (cid: {}, sub: {:?})", client_id, sub.id);
+			    return Ok(());
+			}
                         // give the queue a chance to clear before trying again
                         thread::sleep(Duration::from_millis(100));
                     }

src/server.rs

@@ -25,10 +25,12 @@ use hyper::{
 use rusqlite::OpenFlags;
 use serde::{Deserialize, Serialize};
 use serde_json::json;
+use tokio::sync::Mutex;
 use std::collections::HashMap;
 use std::convert::Infallible;
 use std::net::SocketAddr;
 use std::path::Path;
+use std::sync::Arc;
 use std::sync::atomic::Ordering;
 use std::sync::mpsc::Receiver as MpscReceiver;
 use std::time::Duration;
@@ -54,6 +56,7 @@ async fn handle_web_request(
     broadcast: Sender<Event>,
     event_tx: tokio::sync::mpsc::Sender<SubmittedEvent>,
     shutdown: Receiver<()>,
+    safe_to_read: Arc<Mutex<u64>>,
 ) -> Result<Response<Body>, Infallible> {
     match (
         request.uri().path(),
@@ -114,6 +117,7 @@ async fn handle_web_request(
                                     broadcast,
                                     event_tx,
                                     shutdown,
+				    safe_to_read,
                                 ));
                             }
                             // todo: trace, don't print...
@@ -328,8 +332,13 @@ pub fn start_server(settings: Settings, shutdown_rx: MpscReceiver<()>) -> Result
             2,
             false,
         );
-        db::db_optimize(maintenance_pool.clone()).await;
+
-        db::db_checkpoint(maintenance_pool).await;
+	// Create a mutex that will block readers, so that a
+	// checkpoint can be performed quickly.
+	let safe_to_read = Arc::new(Mutex::new(0));
+
+        db::db_optimize_task(maintenance_pool.clone()).await;
+        db::db_checkpoint_task(maintenance_pool, safe_to_read.clone()).await;
 
         // listen for (external to tokio) shutdown request
         let controlled_shutdown = invoke_shutdown.clone();
@@ -378,6 +387,7 @@ pub fn start_server(settings: Settings, shutdown_rx: MpscReceiver<()>) -> Result
             let event = event_tx.clone();
             let stop = invoke_shutdown.clone();
             let settings = settings.clone();
+            let safe_to_read = safe_to_read.clone();
             async move {
                 // service_fn converts our function into a `Service`
                 Ok::<_, Infallible>(service_fn(move |request: Request<Body>| {
@@ -389,6 +399,7 @@ pub fn start_server(settings: Settings, shutdown_rx: MpscReceiver<()>) -> Result
                         bcast.clone(),
                         event.clone(),
                         stop.subscribe(),
+			safe_to_read.clone(),
                     )
                 }))
             }
@@ -465,6 +476,7 @@ async fn nostr_server(
     broadcast: Sender<Event>,
     event_tx: mpsc::Sender<SubmittedEvent>,
     mut shutdown: Receiver<()>,
+    safe_to_read: Arc<Mutex<u64>>,
 ) {
     // the time this websocket nostr server started
     let orig_start = Instant::now();
@@ -674,8 +686,10 @@ async fn nostr_server(
                                     if let Some(previous_query) = running_queries.insert(s.id.to_owned(), abandon_query_tx) {
                     previous_query.send(()).ok();
                                     }
+		    if s.needs_historical_events() {
                                     // start a database query.  this spawns a blocking database query on a worker thread.
-                                    db::db_query(s, cid.to_owned(), pool.clone(), query_tx.clone(), abandon_query_rx).await;
+					db::db_query(s, cid.to_owned(), pool.clone(), query_tx.clone(), abandon_query_rx,safe_to_read.clone()).await;
+				    }
                 },
                 Err(e) => {
                     info!("Subscription error: {} (cid: {}, sub: {:?})", e, cid, s.id);

src/subscription.rs

@@ -200,6 +200,13 @@ impl Subscription {
     pub fn get_id(&self) -> String {
         self.id.clone()
     }
+
+    /// Determine if any filter is requesting historical (database)
+    /// queries.  If every filter has limit:0, we do not need to query the DB.
+    pub fn needs_historical_events(&self) -> bool {
+	self.filters.iter().any(|f| f.limit!=Some(0))
+    }
+
     /// Determine if this subscription matches a given [`Event`].  Any
     /// individual filter match is sufficient.
     pub fn interested_in_event(&self, event: &Event) -> bool {