Compare commits

..
6 Commits
4 changed files with 18 additions and 12 deletions
Generated
+1 -1
View File
@@ -1550,7 +1550,7 @@ checksum = "38bf9645c8b145698bb0b18a4637dcacbc421ea49bef2317e4fd8065a387cf21"
[[package]] [[package]]
name = "nostr-rs-relay" name = "nostr-rs-relay"
version = "0.8.6" version = "0.8.8"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"async-std", "async-std",
+1 -1
View File
@@ -1,6 +1,6 @@
[package] [package]
name = "nostr-rs-relay" name = "nostr-rs-relay"
version = "0.8.6" version = "0.8.8"
edition = "2021" edition = "2021"
authors = ["Greg Heartsfield <scsibug@imap.cc>"] authors = ["Greg Heartsfield <scsibug@imap.cc>"]
description = "A relay implementation for the Nostr protocol" description = "A relay implementation for the Nostr protocol"
+2 -5
View File
@@ -550,6 +550,7 @@ fn query_from_filter(f: &ReqFilter) -> Option<QueryBuilder<Postgres>> {
let mut query = QueryBuilder::new("SELECT e.\"content\", e.created_at FROM \"event\" e WHERE "); let mut query = QueryBuilder::new("SELECT e.\"content\", e.created_at FROM \"event\" e WHERE ");
// This tracks whether we need to push a prefix AND before adding another clause
let mut push_and = false; let mut push_and = false;
// Query for "authors", allowing prefix matches // Query for "authors", allowing prefix matches
if let Some(auth_vec) = &f.authors { if let Some(auth_vec) = &f.authors {
@@ -747,13 +748,9 @@ fn query_from_filter(f: &ReqFilter) -> Option<QueryBuilder<Postgres>> {
} else { } else {
query.push("e.hidden != 1::bit(1)"); query.push("e.hidden != 1::bit(1)");
} }
// never display expired events // never display expired events
if push_and {
query.push(" AND ");
}
query query
.push("(e.expires_at IS NULL OR e.expires_at > ") .push(" AND (e.expires_at IS NULL OR e.expires_at > ")
.push_bind(Utc.timestamp_opt(utils::unix_time() as i64, 0).unwrap()).push(")"); .push_bind(Utc.timestamp_opt(utils::unix_time() as i64, 0).unwrap()).push(")");
// Apply per-filter limit to this query. // Apply per-filter limit to this query.
+14 -5
View File
@@ -251,8 +251,11 @@ impl SqliteRepo {
impl NostrRepo for SqliteRepo { impl NostrRepo for SqliteRepo {
async fn start(&self) -> Result<()> { async fn start(&self) -> Result<()> {
db_checkpoint_task(self.maint_pool.clone(), Duration::from_secs(60), self.checkpoint_in_progress.clone()).await?; db_checkpoint_task(self.maint_pool.clone(), Duration::from_secs(60),
cleanup_expired(self.maint_pool.clone(), Duration::from_secs(600), self.write_in_progress.clone()).await self.write_in_progress.clone(),
self.checkpoint_in_progress.clone()).await?;
cleanup_expired(self.maint_pool.clone(), Duration::from_secs(600),
self.write_in_progress.clone()).await
} }
async fn migrate_up(&self) -> Result<usize> { async fn migrate_up(&self) -> Result<usize> {
@@ -376,7 +379,7 @@ impl NostrRepo for SqliteRepo {
let mut last_successful_send = Instant::now(); let mut last_successful_send = Instant::now();
// execute the query. // execute the query.
// make the actual SQL query (with parameters inserted) available // make the actual SQL query (with parameters inserted) available
conn.trace(Some(|x| {info!("SQL trace: {:?}", x)})); conn.trace(Some(|x| {trace!("SQL trace: {:?}", x)}));
let mut stmt = conn.prepare_cached(&q)?; let mut stmt = conn.prepare_cached(&q)?;
let mut event_rows = stmt.query(rusqlite::params_from_iter(p))?; let mut event_rows = stmt.query(rusqlite::params_from_iter(p))?;
@@ -510,6 +513,7 @@ impl NostrRepo for SqliteRepo {
let e = hex::decode(event_id).ok(); let e = hex::decode(event_id).ok();
let n = name.to_owned(); let n = name.to_owned();
let mut conn = self.write_pool.get()?; let mut conn = self.write_pool.get()?;
let _write_guard = self.write_in_progress.lock().await;
tokio::task::spawn_blocking(move || { tokio::task::spawn_blocking(move || {
let tx = conn.transaction()?; let tx = conn.transaction()?;
{ {
@@ -537,6 +541,7 @@ impl NostrRepo for SqliteRepo {
/// Update verification timestamp /// Update verification timestamp
async fn update_verification_timestamp(&self, id: u64) -> Result<()> { async fn update_verification_timestamp(&self, id: u64) -> Result<()> {
let mut conn = self.write_pool.get()?; let mut conn = self.write_pool.get()?;
let _write_guard = self.write_in_progress.lock().await;
tokio::task::spawn_blocking(move || { tokio::task::spawn_blocking(move || {
// add some jitter to the verification to prevent everything from stacking up together. // add some jitter to the verification to prevent everything from stacking up together.
let verif_time = now_jitter(600); let verif_time = now_jitter(600);
@@ -559,6 +564,7 @@ impl NostrRepo for SqliteRepo {
/// Update verification record as failed /// Update verification record as failed
async fn fail_verification(&self, id: u64) -> Result<()> { async fn fail_verification(&self, id: u64) -> Result<()> {
let mut conn = self.write_pool.get()?; let mut conn = self.write_pool.get()?;
let _write_guard = self.write_in_progress.lock().await;
tokio::task::spawn_blocking(move || { tokio::task::spawn_blocking(move || {
// add some jitter to the verification to prevent everything from stacking up together. // add some jitter to the verification to prevent everything from stacking up together.
let fail_time = now_jitter(600); let fail_time = now_jitter(600);
@@ -578,6 +584,7 @@ impl NostrRepo for SqliteRepo {
/// Delete verification record /// Delete verification record
async fn delete_verification(&self, id: u64) -> Result<()> { async fn delete_verification(&self, id: u64) -> Result<()> {
let mut conn = self.write_pool.get()?; let mut conn = self.write_pool.get()?;
let _write_guard = self.write_in_progress.lock().await;
tokio::task::spawn_blocking(move || { tokio::task::spawn_blocking(move || {
let tx = conn.transaction()?; let tx = conn.transaction()?;
{ {
@@ -925,7 +932,7 @@ pub fn build_pool(
} }
} }
let manager = if settings.database.in_memory { let manager = if settings.database.in_memory {
SqliteConnectionManager::memory() SqliteConnectionManager::file("file::memory:?cache=shared")
.with_flags(flags) .with_flags(flags)
.with_init(|c| c.execute_batch(STARTUP_SQL)) .with_init(|c| c.execute_batch(STARTUP_SQL))
} else { } else {
@@ -996,7 +1003,7 @@ pub fn delete_expired(conn: &mut PooledConnection) -> Result<usize> {
} }
/// Perform database WAL checkpoint on a regular basis /// Perform database WAL checkpoint on a regular basis
pub async fn db_checkpoint_task(pool: SqlitePool, frequency: Duration, checkpoint_in_progress: Arc<Mutex<u64>>) -> Result<()> { pub async fn db_checkpoint_task(pool: SqlitePool, frequency: Duration, write_in_progress: Arc<Mutex<u64>>, checkpoint_in_progress: Arc<Mutex<u64>>) -> Result<()> {
// TODO; use acquire_many on the reader semaphore to stop them from interrupting this. // TODO; use acquire_many on the reader semaphore to stop them from interrupting this.
tokio::task::spawn(async move { tokio::task::spawn(async move {
// WAL size in pages. // WAL size in pages.
@@ -1011,6 +1018,8 @@ pub async fn db_checkpoint_task(pool: SqlitePool, frequency: Duration, checkpoin
tokio::select! { tokio::select! {
_ = tokio::time::sleep(frequency) => { _ = tokio::time::sleep(frequency) => {
if let Ok(mut conn) = pool.get() { if let Ok(mut conn) = pool.get() {
// block all other writers
let _write_guard = write_in_progress.lock().await;
let mut _guard:Option<MutexGuard<u64>> = None; let mut _guard:Option<MutexGuard<u64>> = None;
// the busy timer will block writers, so don't set // the busy timer will block writers, so don't set
// this any higher than you want max latency for event // this any higher than you want max latency for event