Skip to content

fix: Improve pools and rewards endpoints #784

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
scarmuega merged 1 commit into from
Nov 13, 2025
Merged

fix: Improve pools and rewards endpoints #784

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion crates/minibf/src/routes/accounts.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -566,7 +566,8 @@ where
};

let out = AccountRewardContentInner {
epoch: epoch as i32,
// TODO: This should be handled on write instead of read
epoch: epoch as i32 - 1,
amount: reward.amount.to_string(),
pool_id,
r#type,
Expand Down
199 changes: 128 additions & 71 deletions crates/minibf/src/routes/pools.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
use std::{collections::HashMap, time::Duration};

use axum::{
extract::{Path, Query, State},
http::StatusCode,
Expand All @@ -9,15 +11,15 @@ use blockfrost_openapi::models::{
};
use dolos_cardano::{
model::{AccountState, PoolState},
StakeLog,
FixedNamespace, PoolDelegation, StakeLog,
};
use dolos_core::{BlockSlot, Domain};
use dolos_core::{ArchiveStore, BlockSlot, Domain, EntityKey, TemporalKey};
use itertools::Itertools;
use pallas::{
codec::minicbor,
crypto::hash::Hash,
ledger::{addresses::Network, primitives::StakeCredential},
};
use serde::{Deserialize, Serialize};
use serde_json::json;

use crate::{
Expand All @@ -38,63 +40,28 @@ fn decode_pool_id(pool_id: &str) -> Result<Vec<u8>, Error> {
Err(Error::Code(StatusCode::BAD_REQUEST))
}

struct PoolModelBuilder {
operator: Hash<28>,
state: dolos_cardano::model::PoolState,
}
async fn pool_offchain_metadata(url: &str) -> Option<PoolOffchainMetadata> {
let client = reqwest::Client::builder()
.timeout(Duration::from_secs(10))
.user_agent("Dolos MiniBF")
.build()
.ok()?;

impl IntoModel<PoolListExtendedInner> for PoolModelBuilder {
type SortKey = BlockSlot;
let res = client.get(url).send().await.ok()?;

fn sort_key(&self) -> Option<Self::SortKey> {
Some(self.state.register_slot)
if res.status() != StatusCode::OK {
return None;
}

fn into_model(self) -> Result<PoolListExtendedInner, StatusCode> {
let pool_id = bech32_pool(self.operator)?;

// TODO: implement
let live_stake = "0".to_string();
let active_stake = "0".to_string();

let params = self.state.snapshot.live().map(|x| x.params.clone());

let out = PoolListExtendedInner {
pool_id,
hex: hex::encode(self.operator),
live_stake,
active_stake,
live_saturation: rational_to_f64::<3>(&self.state.live_saturation()),
blocks_minted: self.state.blocks_minted_total as i32,
declared_pledge: params
.as_ref()
.map(|x| x.pledge.to_string())
.unwrap_or_default(),
margin_cost: params
.as_ref()
.map(|x| rational_to_f64::<6>(&x.margin))
.unwrap_or_default(),
fixed_cost: params
.as_ref()
.map(|x| x.cost.to_string())
.unwrap_or_default(),
metadata: params
.as_ref()
.map(|x| {
x.pool_metadata.as_ref().map(|m| {
let out = json!({
"url": m.url,
"hash": m.hash,
});

Box::new(out)
})
})
.unwrap_or_default(),
};
res.json().await.ok()
}

Ok(out)
}
#[derive(Serialize, Deserialize, Clone)]
pub struct PoolOffchainMetadata {
pub name: String,
pub description: String,
pub ticker: String,
pub homepage: String,
}

pub async fn all_extended<D: Domain>(
Expand All @@ -103,39 +70,129 @@ pub async fn all_extended<D: Domain>(
) -> Result<Json<Vec<PoolListExtendedInner>>, Error>
where
Option<PoolState>: From<D::Entity>,
Option<AccountState>: From<D::Entity>,
{
let iter = domain
.iter_cardano_entities::<PoolState>(None)
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

let pagination = Pagination::try_from(params)?;
let chain_summary = domain.get_chain_summary()?;

let mapped: Vec<_> = iter
.into_iter()
let mut live_stake_map = HashMap::new();
for x in domain
.iter_cardano_entities::<AccountState>(None)
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
{
let (_, state) = x.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
if let Some(PoolDelegation::Pool(hash)) = state.pool.live() {
let stake = state.stake.live().map(|x| x.total()).unwrap_or(0);
live_stake_map
.entry(*hash)
.and_modify(|entry| *entry += stake)
.or_insert(stake);
};
}
Comment on lines +79 to +91

@coderabbitai coderabbitai Bot Nov 12, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Critical performance concern: iterating all accounts on every request.

The function iterates over all AccountState entities to build the live_stake_map on every call to all_extended. With a large number of accounts, this will:

  • Consume significant memory by loading all account states
  • Take substantial time to process, blocking the request
  • Scale poorly as the number of accounts grows

This is particularly problematic since this endpoint will be called frequently with different pagination parameters.

Consider these alternatives:

  1. Maintain a materialized aggregate: Pre-compute and store pool live stakes during write operations, updating incrementally as delegations change
  2. Use a database query: If using a database, leverage SQL aggregation to compute stakes efficiently
  3. Cache the live stake map: Cache the computed map with a TTL, refreshing periodically rather than on every request

Would you like me to help design a caching or materialization strategy?

🤖 Prompt for AI Agents 
crates/minibf/src/routes/pools.rs around lines 79-91: the code iterates all
AccountState entities on every request to build live_stake_map which is O(N) and
will not scale; replace this hot-path work with one of the following fixes: (1)
read precomputed per-pool live stake aggregates maintained/updated on write
operations (recommended) by updating the aggregate whenever an account's
delegation or stake changes, (2) perform a single database-side aggregation
query that groups by pool and sums stakes instead of loading each AccountState
into memory, or (3) add an in-memory cached live_stake_map with a TTL or
background refresh task and serve requests from the cache; implement one of
these approaches and remove the per-request full iteration so the endpoint
responds from the aggregate/cache/query rather than scanning all accounts.

let circulating_supply = dolos_cardano::load_epoch::<D>(domain.state())
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.initial_pots
.circulating();
let optimal = domain
.get_current_effective_pparams()?
.ensure_k()
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

let pools = domain
.iter_cardano_entities::<PoolState>(None)
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.flat_map(|x| {
let Ok((key, state)) = x else {
return Some(Err(StatusCode::INTERNAL_SERVER_ERROR));
};

if state.snapshot.live().map(|x| x.is_retired).unwrap_or(false) {
return None;
}

let operator = Hash::<28>::from(key);

let builder = PoolModelBuilder { operator, state };

Some(Ok(builder.into_model_with_sort_key()))
Some(Ok((state.register_slot, (key, state))))
})
.collect::<Result<Result<Vec<(BlockSlot, PoolListExtendedInner)>, _>, StatusCode>>()??
.collect::<Result<Vec<(BlockSlot, (EntityKey, PoolState))>, StatusCode>>()?
.into_iter()
.sorted_by(|a, b| Ord::cmp(&a.0, &b.0))
.map(|(_, x)| x)
.skip(pagination.skip())
.take(pagination.count)
.collect();
.collect_vec();

let mut out = vec![];
for (key, pool) in pools {
let poolhex = hex::encode(pool.operator);
let pool_id = bech32_pool(pool.operator)?;
let params = pool.snapshot.live().map(|x| x.params.clone());
let metadata = match params.as_ref() {
Some(x) => match x.pool_metadata.as_ref() {
Some(y) => {
let out = match pool_offchain_metadata(&y.url).await {
Some(meta) => json!({
"url": y.url,
"hash": y.hash,
"ticker": meta.ticker,
"name": meta.name,
"description": meta.description,
"homepage": meta.homepage
}),
None => json!({
"url": y.url,
"hash": y.hash,
"ticker": None::<String>,
"name": None::<String>,
"description": None::<String>,
"homepage": None::<String>
}),
};

Some(Box::new(out))
}
None => None,
},
None => None,
};

Ok(Json(mapped))
// Fetch live and active stake logs
let live = live_stake_map.get(&pool.operator).copied();
let Some(epoch) = pool.snapshot.epoch() else {
return Err(StatusCode::INTERNAL_SERVER_ERROR.into());
};
let active_slot = chain_summary.epoch_start(epoch - 1);
let Ok(active) = domain.archive().read_log_typed::<StakeLog>(
StakeLog::NS,
&(TemporalKey::from(active_slot), key.clone()).into(),
) else {
return Err(StatusCode::INTERNAL_SERVER_ERROR.into());
};

out.push(PoolListExtendedInner {
pool_id,
hex: poolhex,
live_stake: live.map(|x| x.to_string()).unwrap_or("0".to_string()),
active_stake: active
.map(|x| x.total_stake.to_string())
.unwrap_or("0".to_string()),
live_saturation: live
.map(|x| x as f64 * optimal as f64 / circulating_supply as f64)
.unwrap_or_default(),
blocks_minted: pool.blocks_minted_total as i32,
declared_pledge: params
.as_ref()
.map(|x| x.pledge.to_string())
.unwrap_or_default(),
margin_cost: params
.as_ref()
.map(|x| rational_to_f64::<6>(&x.margin))
.unwrap_or_default(),
fixed_cost: params
.as_ref()
.map(|x| x.cost.to_string())
.unwrap_or_default(),
metadata,
});
}
Comment on lines +122 to +193

@coderabbitai coderabbitai Bot Nov 12, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

Sequential HTTP requests will cause severe performance degradation.

Inside the main loop, pool_offchain_metadata (line 129) makes an HTTP request for each pool sequentially. If the pagination returns 100 pools, this performs 100 sequential HTTP requests, each with a 10-second timeout. This could take minutes to complete and severely degrade user experience.

Apply this refactor to fetch metadata concurrently:

+use futures::future::join_all;
+
 pub async fn all_extended<D: Domain>(
     Query(params): Query<PaginationParameters>,
     State(domain): State<Facade<D>>,
 ) -> Result<Json<Vec<PoolListExtendedInner>>, Error>
 where
     Option<PoolState>: From<D::Entity>,
     Option<AccountState>: From<D::Entity>,
 {
     // ... existing code to build pools vec ...
     
+    // Fetch all metadata concurrently
+    let metadata_futures: Vec<_> = pools
+        .iter()
+        .map(|(_, pool)| {
+            let url = pool
+                .snapshot
+                .live()
+                .and_then(|x| x.params.as_ref())
+                .and_then(|x| x.pool_metadata.as_ref())
+                .map(|x| x.url.clone());
+            async move {
+                match url {
+                    Some(u) => pool_offchain_metadata(&u).await,
+                    None => None,
+                }
+            }
+        })
+        .collect();
+    
+    let metadata_results = join_all(metadata_futures).await;
     
     let mut out = vec![];
-    for (key, pool) in pools {
+    for ((key, pool), fetched_metadata) in pools.into_iter().zip(metadata_results) {
         let poolhex = hex::encode(pool.operator);
         let pool_id = bech32_pool(pool.operator)?;
         let params = pool.snapshot.live().map(|x| x.params.clone());
         let metadata = match params.as_ref() {
             Some(x) => match x.pool_metadata.as_ref() {
                 Some(y) => {
-                    let out = match pool_offchain_metadata(&y.url).await {
+                    let out = match fetched_metadata {
                         Some(meta) => json!({
                             "url": y.url,
                             "hash": y.hash,
                             "ticker": meta.ticker,
                             // ... rest of fields

This fetches all metadata concurrently, dramatically reducing response time.

Committable suggestion skipped: line range outside the PR's diff.

🤖 Prompt for AI Agents 
In crates/minibf/src/routes/pools.rs around lines 122 to 193, the code calls
pool_offchain_metadata sequentially inside the loop causing N HTTP requests to
run one-by-one; instead collect all pool metadata URLs up front, spawn
concurrent tasks (e.g., FuturesUnordered or join_all) to fetch
pool_offchain_metadata for each URL with proper timeout/error handling, await
all results, build a map from operator (or URL) to fetched metadata, and then
iterate the pools to construct PoolListExtendedInner using the pre-fetched
metadata rather than calling the HTTP function inline.


Ok(Json(out))
}

struct PoolDelegatorModelBuilder {
Expand Down
Loading