opt(torii-core): move off queryqueue for executing tx

bin/torii/src/main.rs

@@ -31,6 +31,7 @@ use tokio::sync::broadcast;
 use tokio::sync::broadcast::Sender;
 use tokio_stream::StreamExt;
 use torii_core::engine::{Engine, EngineConfig, IndexingFlags, Processors};
+use torii_core::executor::Executor;
 use torii_core::processors::event_message::EventMessageProcessor;
 use torii_core::processors::generate_event_processors_map;
 use torii_core::processors::metadata_update::MetadataUpdateProcessor;
@@ -185,7 +186,8 @@ async fn main() -> anyhow::Result<()> {
     // Get world address
     let world = WorldContractReader::new(args.world_address, provider.clone());
 
-    let db = Sql::new(pool.clone(), args.world_address).await?;
+    let (mut executor, sender) = Executor::new(pool.clone()).await?;
+    let db = Sql::new(pool.clone(), args.world_address, sender.clone()).await?;
 
     let processors = Processors {
         event: generate_event_processors_map(vec![
@@ -289,6 +291,7 @@ async fn main() -> anyhow::Result<()> {
 
     tokio::select! {
         res = engine.start() => res?,
+        _ = executor.run() => {},
         _ = proxy_server.start(shutdown_tx.subscribe()) => {},
         _ = graphql_server => {},
         _ = grpc_server => {},

crates/torii/core/src/engine.rs

@@ -21,6 +21,7 @@ use tokio::task::JoinSet;
 use tokio::time::{sleep, Instant};
 use tracing::{debug, error, info, trace, warn};
 
+use crate::executor::{QueryMessage, QueryType};
 use crate::processors::event_message::EventMessageProcessor;
 use crate::processors::{BlockProcessor, EventProcessor, TransactionProcessor};
 use crate::sql::Sql;
@@ -151,7 +152,7 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
         // use the start block provided by user if head is 0
         let (head, _, _) = self.db.head().await?;
         if head == 0 {
-            self.db.set_head(self.config.start_block);
+            self.db.set_head(self.config.start_block)?;
         } else if self.config.start_block != 0 {
             warn!(target: LOG_TARGET, "Start block ignored, stored head exists and will be used instead.");
         }
@@ -179,7 +180,7 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
                             }
 
                             match self.process(fetch_result).await {
-                                Ok(()) => {}
+                                Ok(()) => self.db.execute()?,
                                 Err(e) => {
                                     error!(target: LOG_TARGET, error = %e, "Processing fetched data.");
                                     erroring_out = true;
@@ -407,15 +408,14 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
                             // provider. So we can fail silently and try
                             // again in the next iteration.
                             warn!(target: LOG_TARGET, transaction_hash = %format!("{:#x}", transaction_hash), "Retrieving pending transaction receipt.");
-                            self.db.set_head(data.block_number - 1);
+                            self.db.set_head(data.block_number - 1)?;
                             if let Some(tx) = last_pending_block_tx {
-                                self.db.set_last_pending_block_tx(Some(tx));
+                                self.db.set_last_pending_block_tx(Some(tx))?;
                             }
 
                             if let Some(tx) = last_pending_block_world_tx {
-                                self.db.set_last_pending_block_world_tx(Some(tx));
+                                self.db.set_last_pending_block_world_tx(Some(tx))?;
                             }
-                            self.db.execute().await?;
                             return Ok(());
                         }
                         _ => {
@@ -441,18 +441,16 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
 
         // Set the head to the last processed pending transaction
         // Head block number should still be latest block number
-        self.db.set_head(data.block_number - 1);
+        self.db.set_head(data.block_number - 1)?;
 
         if let Some(tx) = last_pending_block_tx {
-            self.db.set_last_pending_block_tx(Some(tx));
+            self.db.set_last_pending_block_tx(Some(tx))?;
         }
 
         if let Some(tx) = last_pending_block_world_tx {
-            self.db.set_last_pending_block_world_tx(Some(tx));
+            self.db.set_last_pending_block_world_tx(Some(tx))?;
         }
 
-        self.db.execute().await?;
-
         Ok(())
     }
 
@@ -486,20 +484,14 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
                 self.process_block(block_number, data.blocks[&block_number]).await?;
                 last_block = block_number;
             }
-
-            if self.db.query_queue.queue.len() >= QUERY_QUEUE_BATCH_SIZE {
-                self.db.execute().await?;
-            }
         }
 
         // Process parallelized events
         self.process_tasks().await?;
 
-        self.db.set_head(data.latest_block_number);
-        self.db.set_last_pending_block_world_tx(None);
-        self.db.set_last_pending_block_tx(None);
-
-        self.db.execute().await?;
+        self.db.set_head(data.latest_block_number)?;
+        self.db.set_last_pending_block_world_tx(None)?;
+        self.db.set_last_pending_block_tx(None)?;
 
         Ok(())
     }
@@ -536,10 +528,7 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
         }
 
         // Join all tasks
-        while let Some(result) = set.join_next().await {
-            let local_db = result??;
-            self.db.merge(local_db)?;
-        }
+        while let Some(_) = set.join_next().await {}
 
         Ok(())
     }
@@ -688,7 +677,7 @@ impl<P: Provider + Send + Sync + std::fmt::Debug + 'static> Engine<P> {
         transaction_hash: Felt,
     ) -> Result<()> {
         if self.config.flags.contains(IndexingFlags::RAW_EVENTS) {
-            self.db.store_event(event_id, event, transaction_hash, block_timestamp);
+            self.db.store_event(event_id, event, transaction_hash, block_timestamp)?;
         }
 
         let event_key = event.keys[0];

crates/torii/core/src/query_queue.rs → crates/torii/core/src/executor.rs

@@ -1,9 +1,11 @@
 use std::collections::VecDeque;
+use std::mem;
 
 use anyhow::{Context, Result};
 use dojo_types::schema::{Struct, Ty};
-use sqlx::{FromRow, Pool, Sqlite};
+use sqlx::{FromRow, Pool, Sqlite, Transaction};
 use starknet::core::types::Felt;
+use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
 
 use crate::simple_broker::SimpleBroker;
 use crate::types::{
@@ -28,12 +30,6 @@ pub enum BrokerMessage {
     EventEmitted(EventEmitted),
 }
 
-#[derive(Debug, Clone)]
-pub struct QueryQueue {
-    pool: Pool<Sqlite>,
-    pub queue: VecDeque<(String, Vec<Argument>, QueryType)>,
-}
-
 #[derive(Debug, Clone)]
 pub struct DeleteEntityQuery {
     pub entity_id: String,
@@ -49,30 +45,36 @@ pub enum QueryType {
     EventMessage(Ty),
     RegisterModel,
     StoreEvent,
+    Execute,
     Other,
 }
 
-impl QueryQueue {
-    pub fn new(pool: Pool<Sqlite>) -> Self {
-        QueryQueue { pool, queue: VecDeque::new() }
-    }
+pub struct Executor<'c> {
+    pool: Pool<Sqlite>,
+    transaction: Transaction<'c, Sqlite>,
+    publish_queue: VecDeque<BrokerMessage>,
+    rx: UnboundedReceiver<QueryMessage>,
+}
 
-    pub fn enqueue<S: Into<String>>(
-        &mut self,
-        statement: S,
-        arguments: Vec<Argument>,
-        query_type: QueryType,
-    ) {
-        self.queue.push_back((statement.into(), arguments, query_type));
-    }
+pub struct QueryMessage {
+    pub statement: String,
+    pub arguments: Vec<Argument>,
+    pub query_type: QueryType,
+}
+
+impl<'c> Executor<'c> {
+    pub async fn new(pool: Pool<Sqlite>) -> Result<(Self, UnboundedSender<QueryMessage>)> {
+        let (tx, rx) = unbounded_channel();
+        let transaction = pool.begin().await?;
+        let publish_queue = VecDeque::new();
 
-    pub async fn execute_all(&mut self) -> Result<()> {
-        let mut tx = self.pool.begin().await?;
-        // publishes that are related to queries in the queue, they should be sent
-        // after the queries are executed
-        let mut publish_queue = VecDeque::new();
+        Ok((Executor { pool, transaction, publish_queue, rx }, tx))
+    }
 
-        while let Some((statement, arguments, query_type)) = self.queue.pop_front() {
+    pub async fn run(&mut self) -> Result<()> {
+        while let Some(msg) = self.rx.recv().await {
+            let tx = &mut self.transaction;
+            let QueryMessage { statement, arguments, query_type } = msg;
             let mut query = sqlx::query(&statement);
 
             for arg in &arguments {
@@ -87,17 +89,17 @@ impl QueryQueue {
 
             match query_type {
                 QueryType::SetEntity(entity) => {
-                    let row = query.fetch_one(&mut *tx).await.with_context(|| {
+                    let row = query.fetch_one(&mut **tx).await.with_context(|| {
                         format!("Failed to execute query: {:?}, args: {:?}", statement, arguments)
                     })?;
                     let mut entity_updated = EntityUpdated::from_row(&row)?;
                     entity_updated.updated_model = Some(entity);
                     entity_updated.deleted = false;
                     let broker_message = BrokerMessage::EntityUpdated(entity_updated);
-                    publish_queue.push_back(broker_message);
+                    self.publish_queue.push_back(broker_message);
                 }
                 QueryType::DeleteEntity(entity) => {
-                    let delete_model = query.execute(&mut *tx).await.with_context(|| {
+                    let delete_model = query.execute(&mut **tx).await.with_context(|| {
                         format!("Failed to execute query: {:?}, args: {:?}", statement, arguments)
                     })?;
                     if delete_model.rows_affected() == 0 {
@@ -111,7 +113,7 @@ impl QueryQueue {
                     .bind(entity.block_timestamp)
                     .bind(entity.event_id)
                     .bind(entity.entity_id)
-                    .fetch_one(&mut *tx)
+                    .fetch_one(&mut **tx)
                     .await?;
                     let mut entity_updated = EntityUpdated::from_row(&row)?;
                     entity_updated.updated_model =
@@ -121,55 +123,63 @@ impl QueryQueue {
                         "SELECT count(*) FROM entity_model WHERE entity_id = ?",
                     )
                     .bind(entity_updated.id.clone())
-                    .fetch_one(&mut *tx)
+                    .fetch_one(&mut **tx)
                     .await?;
 
                     // Delete entity if all of its models are deleted
                     if count == 0 {
                         sqlx::query("DELETE FROM entities WHERE id = ?")
                             .bind(entity_updated.id.clone())
-                            .execute(&mut *tx)
+                            .execute(&mut **tx)
                             .await?;
                         entity_updated.deleted = true;
                     }
 
                     let broker_message = BrokerMessage::EntityUpdated(entity_updated);
-                    publish_queue.push_back(broker_message);
+                    self.publish_queue.push_back(broker_message);
                 }
                 QueryType::RegisterModel => {
-                    let row = query.fetch_one(&mut *tx).await.with_context(|| {
+                    let row = query.fetch_one(&mut **tx).await.with_context(|| {
                         format!("Failed to execute query: {:?}, args: {:?}", statement, arguments)
                     })?;
                     let model_registered = ModelRegistered::from_row(&row)?;
-                    publish_queue.push_back(BrokerMessage::ModelRegistered(model_registered));
+                    self.publish_queue.push_back(BrokerMessage::ModelRegistered(model_registered));
                 }
                 QueryType::EventMessage(entity) => {
-                    let row = query.fetch_one(&mut *tx).await.with_context(|| {
+                    let row = query.fetch_one(&mut **tx).await.with_context(|| {
                         format!("Failed to execute query: {:?}, args: {:?}", statement, arguments)
                     })?;
                     let mut event_message = EventMessageUpdated::from_row(&row)?;
                     event_message.updated_model = Some(entity);
                     let broker_message = BrokerMessage::EventMessageUpdated(event_message);
-                    publish_queue.push_back(broker_message);
+                    self.publish_queue.push_back(broker_message);
                 }
                 QueryType::StoreEvent => {
-                    let row = query.fetch_one(&mut *tx).await.with_context(|| {
+                    let row = query.fetch_one(&mut **tx).await.with_context(|| {
                         format!("Failed to execute query: {:?}, args: {:?}", statement, arguments)
                     })?;
                     let event = EventEmitted::from_row(&row)?;
-                    publish_queue.push_back(BrokerMessage::EventEmitted(event));
+                    self.publish_queue.push_back(BrokerMessage::EventEmitted(event));
+                }
+                QueryType::Execute => {
+                    self.execute().await?;
                 }
                 QueryType::Other => {
-                    query.execute(&mut *tx).await.with_context(|| {
+                    query.execute(&mut **tx).await.with_context(|| {
                         format!("Failed to execute query: {:?}, args: {:?}", statement, arguments)
                     })?;
                 }
             }
         }
 
-        tx.commit().await?;
+        Ok(())
+    }
+
+    pub async fn execute(&mut self) -> Result<()> {
+        let transaction = mem::replace(&mut self.transaction, self.pool.begin().await?);
+        transaction.commit().await?;
 
-        while let Some(message) = publish_queue.pop_front() {
+        while let Some(message) = self.publish_queue.pop_front() {
             send_broker_message(message);
         }
 

crates/torii/core/src/lib.rs

@@ -1,9 +1,9 @@
 pub mod cache;
 pub mod engine;
 pub mod error;
+pub mod executor;
 pub mod model;
 pub mod processors;
-pub mod query_queue;
 pub mod simple_broker;
 pub mod sql;
 pub mod types;

crates/torii/core/src/processors/metadata_update.rs

@@ -64,7 +64,7 @@ where
             uri = %uri_str,
             "Resource metadata set."
         );
-        db.set_metadata(resource, &uri_str, block_timestamp);
+        db.set_metadata(resource, &uri_str, block_timestamp)?;
 
         let db = db.clone();
         let resource = *resource;
@@ -83,9 +83,7 @@ where
 async fn try_retrieve(mut db: Sql, resource: Felt, uri_str: String) {
     match metadata(uri_str.clone()).await {
         Ok((metadata, icon_img, cover_img)) => {
-            db.update_metadata(&resource, &uri_str, &metadata, &icon_img, &cover_img)
-                .await
-                .unwrap();
+            db.update_metadata(&resource, &uri_str, &metadata, &icon_img, &cover_img).unwrap();
             info!(
                 target: LOG_TARGET,
                 resource = %format!("{:#x}", resource),

crates/torii/core/src/processors/store_transaction.rs

@@ -21,7 +21,7 @@ impl<P: Provider + Sync + std::fmt::Debug> TransactionProcessor<P> for StoreTran
         transaction: &Transaction,
     ) -> Result<(), Error> {
         let transaction_id = format!("{:#064x}:{:#x}", block_number, transaction_hash);
-        db.store_transaction(transaction, &transaction_id, block_timestamp);
+        db.store_transaction(transaction, &transaction_id, block_timestamp)?;
         Ok(())
     }
 }