Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2be75e18fb | ||
|
|
5f6ff4c2b7 | ||
|
|
df411c24fb | ||
|
|
39f9984c4f | ||
|
|
9d55731073 | ||
|
|
5638f70d66 |
Generated
+1
-1
@@ -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
@@ -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"
|
||||||
|
|||||||
@@ -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
@@ -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
|
||||||
|
|||||||
Reference in New Issue
Block a user