Implement sorting algorithm (#25)

* feat: updated priority to queue

* feat: implemented priority config

* feat: added priority module

* chore: adjusted lint

* feat: implemented priority quote

* feat: implemented storage query to be used in priority module

* test: implemented priority test

* chore: removed dbg

* chore: added validation when the state is empty

* docs: added priority docs

* chore: updated config structure

* docs: updated docs

docs/app/_meta.tsx

@@ -2,5 +2,6 @@ export default {
   index: 'Introduction',
   installation: 'Installation',
   configuration: 'Configuration',
+  priority: 'Priority Queues',
   design: 'Design',
 };

docs/app/configuration/page.mdx

@@ -36,6 +36,11 @@ uri = "https://mainnet.utxorpc-v0.demeter.run"
 # metadata is optional
 [u5c.metadata]
 "dmtr-api-key" = "your key"
+
+# optional config
+[[queues]]
+name = "banana"
+weight = 2
 ```
 
 ### `storage` section
@@ -102,6 +107,18 @@ The `u5c.metadata` section is optional and defines metadata for U5C connection,
 
 - `key`: a string value.
 
+### `queues` section
+
+The `queues` section defines the options to create queues.
+
+| property | type    | example    |
+| -------- | ------- | ---------- |
+| name     | string  | queue_name |
+| weight   | number  | 2          |
+
+- `name`: queue name that can be used in sending transactions.
+- `weight`: the weight of the queue. A queue with a bigger weight will have more priority to process transactions.
+
 ## Tips
 
 For U5C, the UTxO RPC from [Demeter](https://demeter.run/ports/cardano-utxorpc) can be used.

docs/app/design/adrs/004-priority/page.md → docs/app/design/adrs/005-priority/page.md

@@ -23,6 +23,6 @@ We chose **Queue** because it allows the user to define a configurable queue, an
 
 **Rationale**
   - Supports configurable queues.
-  - Quote of transactions.
+  - Quota of transactions.
   - Handles when a transaction can be moved to the fanout stage.
 

docs/app/design/page.mdx

@@ -77,7 +77,8 @@ erDiagram
         TEXT id PK
         BLOB raw
         TEXT status
-        INTEGER priority
+        INTEGER slot
+        TEXT queue
         DATETIME created_at
         DATETIME updated_at
     }

docs/app/priority/page.mdx

@@ -0,0 +1,16 @@
+---
+title: Priority Queues
+sidebarTitle: Priority Queues
+---
+
+## Priority Queues
+
+With the Priority module, Boros allows users to configure queues to prioritise transactions. Each queue has a weight that Boros uses to calculate how many slots of the quota a queue can use. For transactions where there's no queue, a default queue is set. When a queue is removed from the file and Boros still has transactions pending for the queue removed, Boros uses the default queue weight to process unknown queues.
+
+Follow the [configuration](../configuration) to configure the priority queues.
+
+### Example
+The quota is 10, so the fanout stage has a backpressure of 10, meaning only 10 transactions can be waiting for propagation. The quota is released when transactions are sent, allowing new transactions to wait. Furthermore, two queues are configured: Queue A with a weight of 3 and Queue B with a weight of 2. Queue A has higher priority than Queue B, so Queue A receives 6 slots and Queue B receives 4 slots from the quota.
+
+### Default
+Priority configurations are not required. If not configured, the default queue is used. The default queue has a weight of 1 but can be replaced by setting a configuration with the name default. Transactions with unknown queues are processed using the default weight.

examples/config.toml

@@ -22,3 +22,8 @@ uri = "https://mainnet.utxorpc-v0.demeter.run"
 # metadata is optional
 [u5c.metadata]
 "dmtr-api-key" = "your key"
+
+[[queues]]
+name = "banana"
+weight = 2
+

src/main.rs

@@ -1,7 +1,8 @@
-use std::{env, error::Error, path, sync::Arc};
+use std::{collections::HashSet, env, error::Error, path, sync::Arc};
 
 use anyhow::Result;
 use dotenv::dotenv;
+use priority::DEFAULT_QUEUE;
 use serde::Deserialize;
 use storage::sqlite::{SqliteCursor, SqliteStorage, SqliteTransaction};
 use tokio::try_join;
@@ -10,6 +11,7 @@ use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt, Env
 
 mod ledger;
 mod pipeline;
+mod priority;
 mod server;
 mod storage;
 
@@ -49,12 +51,14 @@ struct Config {
     storage: storage::Config,
     peer_manager: pipeline::fanout::PeerManagerConfig,
     monitor: pipeline::monitor::Config,
+    #[serde(default)]
+    queues: HashSet<priority::QueueConfig>,
     u5c: ledger::u5c::Config,
 }
 
 impl Config {
     pub fn new() -> Result<Self, Box<dyn Error>> {
-        let config = config::Config::builder()
+        let mut config: Config = config::Config::builder()
             .add_source(
                 config::File::with_name(&env::var("BOROS_CONFIG").unwrap_or("boros.toml".into()))
                     .required(false),
@@ -64,6 +68,9 @@ impl Config {
             .build()?
             .try_deserialize()?;
 
+        (!config.queues.iter().any(|q| q.name == DEFAULT_QUEUE))
+            .then(|| config.queues.insert(Default::default()));
+
         Ok(config)
     }
 }

src/pipeline/fanout/mod.rs

@@ -11,6 +11,7 @@ use tracing::info;
 
 use crate::{
     ledger::{relay::RelayDataAdapter, u5c::U5cDataAdapter},
+    priority::Priority,
     storage::{sqlite::SqliteTransaction, Transaction, TransactionStatus},
 };
 
@@ -31,7 +32,7 @@ pub enum FanoutError {
 }
 
 pub enum FanoutUnit {
-    Transaction(Transaction),
+    Transactions(Vec<Transaction>),
     PeerDiscovery(String),
 }
 
@@ -42,6 +43,7 @@ pub struct Stage {
     relay_adapter: Arc<dyn RelayDataAdapter + Send + Sync>,
     u5c_adapter: Arc<dyn U5cDataAdapter>,
     storage: Arc<SqliteTransaction>,
+    priority: Arc<Priority>,
 }
 
 impl Stage {
@@ -50,12 +52,14 @@ impl Stage {
         relay_adapter: Arc<dyn RelayDataAdapter + Send + Sync>,
         u5c_adapter: Arc<dyn U5cDataAdapter>,
         storage: Arc<SqliteTransaction>,
+        priority: Arc<Priority>,
     ) -> Self {
         Self {
             config,
             relay_adapter,
             u5c_adapter,
             storage,
+            priority,
         }
     }
 }
@@ -94,14 +98,14 @@ impl gasket::framework::Worker<Stage> for Worker {
         let additional_peers_required =
             desired_count as usize - self.peer_manager.connected_peers_count().await;
 
-        if let Some(tx) = stage
-            .storage
+        let transactions = stage
+            .priority
             .next(TransactionStatus::Validated)
             .await
-            .or_retry()?
-        {
-            info!("Found Transaction: {}", tx.id);
-            return Ok(WorkSchedule::Unit(FanoutUnit::Transaction(tx)));
+            .or_retry()?;
+
+        if !transactions.is_empty() {
+            return Ok(WorkSchedule::Unit(FanoutUnit::Transactions(transactions)));
         }
 
         if self.peer_discovery_queue < additional_peers_required as u8 {
@@ -135,18 +139,20 @@ impl gasket::framework::Worker<Stage> for Worker {
 
     async fn execute(&mut self, unit: &FanoutUnit, stage: &mut Stage) -> Result<(), WorkerError> {
         match unit {
-            FanoutUnit::Transaction(tx) => {
-                let mut transaction = tx.clone();
-                info!("Propagating Transaction: {}", transaction.id);
+            FanoutUnit::Transactions(transactions) => {
+                for transaction in transactions {
+                    let mut transaction = transaction.clone();
+                    info!("Propagating Transaction: {}", transaction.id);
 
-                let tip = stage.u5c_adapter.fetch_tip().await.or_retry()?;
+                    let tip = stage.u5c_adapter.fetch_tip().await.or_retry()?;
 
-                self.peer_manager.add_tx(transaction.raw.clone()).await;
+                    self.peer_manager.add_tx(transaction.raw.clone()).await;
 
-                transaction.status = TransactionStatus::InFlight;
-                transaction.slot = Some(tip.0);
+                    transaction.status = TransactionStatus::InFlight;
+                    transaction.slot = Some(tip.0);
 
-                stage.storage.update(&transaction).await.or_retry()?;
+                    stage.storage.update(&transaction).await.or_retry()?;
+                }
             }
             FanoutUnit::PeerDiscovery(peer_addr) => {
                 info!("Connecting to peer: {}", peer_addr);

src/pipeline/ingest.rs

@@ -4,17 +4,21 @@ use gasket::framework::*;
 use tokio::time::sleep;
 use tracing::info;
 
-use crate::storage::{sqlite::SqliteTransaction, Transaction, TransactionStatus};
+use crate::{
+    priority::Priority,
+    storage::{sqlite::SqliteTransaction, Transaction, TransactionStatus},
+};
 
 #[derive(Stage)]
-#[stage(name = "ingest", unit = "Transaction", worker = "Worker")]
+#[stage(name = "ingest", unit = "Vec<Transaction>", worker = "Worker")]
 pub struct Stage {
     storage: Arc<SqliteTransaction>,
+    priority: Arc<Priority>,
 }
 
 impl Stage {
-    pub fn new(storage: Arc<SqliteTransaction>) -> Self {
-        Self { storage }
+    pub fn new(storage: Arc<SqliteTransaction>, priority: Arc<Priority>) -> Self {
+        Self { storage, priority }
     }
 }
 
@@ -29,27 +33,38 @@ impl gasket::framework::Worker<Stage> for Worker {
     async fn schedule(
         &mut self,
         stage: &mut Stage,
-    ) -> Result<WorkSchedule<Transaction>, WorkerError> {
-        if let Some(tx) = stage
-            .storage
+    ) -> Result<WorkSchedule<Vec<Transaction>>, WorkerError> {
+        let transactions = stage
+            .priority
             .next(TransactionStatus::Pending)
             .await
-            .or_retry()?
-        {
-            return Ok(WorkSchedule::Unit(tx));
+            .or_retry()?;
+
+        if !transactions.is_empty() {
+            return Ok(WorkSchedule::Unit(transactions));
         }
 
         sleep(Duration::from_secs(1)).await;
         Ok(WorkSchedule::Idle)
     }
 
-    async fn execute(&mut self, unit: &Transaction, stage: &mut Stage) -> Result<(), WorkerError> {
-        let mut transaction = unit.clone();
+    async fn execute(
+        &mut self,
+        unit: &Vec<Transaction>,
+        stage: &mut Stage,
+    ) -> Result<(), WorkerError> {
+        info!("validating {} transactions", unit.len());
 
-        info!("ingest {}", transaction.id);
+        let transactions = unit
+            .iter()
+            .map(|tx| {
+                let mut tx = tx.clone();
+                tx.status = TransactionStatus::Validated;
+                tx
+            })
+            .collect();
 
-        transaction.status = TransactionStatus::Validated;
-        stage.storage.update(&transaction).await.or_retry()?;
+        stage.storage.update_batch(&transactions).await.or_retry()?;
 
         Ok(())
     }

src/pipeline/mod.rs

@@ -7,6 +7,7 @@ use crate::{
         relay::{MockRelayDataAdapter, RelayDataAdapter},
         u5c::{Point, U5cDataAdapterImpl},
     },
+    priority::Priority,
     storage::{
         sqlite::{SqliteCursor, SqliteTransaction},
         Cursor,
@@ -28,12 +29,15 @@ pub async fn run(
         Arc::new(MockRelayDataAdapter::new());
     let u5c_data_adapter = Arc::new(U5cDataAdapterImpl::try_new(config.u5c, cursor).await?);
 
-    let ingest = ingest::Stage::new(tx_storage.clone());
+    let priority = Arc::new(Priority::new(tx_storage.clone(), config.queues));
+
+    let ingest = ingest::Stage::new(tx_storage.clone(), priority.clone());
     let fanout = fanout::Stage::new(
         config.peer_manager,
         relay_adapter.clone(),
         u5c_data_adapter.clone(),
         tx_storage.clone(),
+        priority.clone(),
     );
 
     let monitor = monitor::Stage::new(