BitcoinCash/riftenlabs-indexer
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Better

Browse source
This commit is contained in:
jakobsn 2026-08-25 12:17:34 +02:00
parent 87a5187676
commit 852e389ed1
12 changed files with 110 additions and 285 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
src/db/cauldron/candlestick/tests.rs
View file

@ -122,7 +122,6 @@ async fn insert_tx(
Some(ts),
leg.deltas.0,
leg.deltas.1,
None,
)
.await
.unwrap();

13
src/db/cauldron/fees.rs
View file

@ -47,20 +47,13 @@ const SQRT_SCALE: u64 = 1_000_000_000;
/// What happened between two consecutive pool states.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(i64)]
pub enum StepKind {
/// Not a step we can read: a reserve is missing, zero, or unchanged.
Unreadable = 0,
Unreadable,
/// Both reserves moved the same way — a deposit or a withdrawal.
LiquidityChange = 1,
LiquidityChange,
/// The reserves moved in opposite directions — a swap.
Trade = 2,
}
impl StepKind {
pub fn as_i64(self) -> i64 {
self as i64
}
Trade,
}
/// A pool's reserves at one point in time.

339
src/db/cauldron/pool.rs
View file

@ -16,7 +16,7 @@ use crate::db::cauldron::fees::{
use anyhow::{Context, Result};
use bitcoincash::TokenID;
use log::{debug, error, info, warn};
use log::{debug, info, warn};
use malachite::Integer;
use riftenlabs_defi::{cauldron::ParsedContract, chainutil::OutPointHash};
use rust_decimal::prelude::Zero;
@ -59,13 +59,7 @@ pub async fn create_table(pool: &SqlitePool) {
sats BIGINT NOT NULL,
token_amount BIGINT NOT NULL,
sats_delta BIGINT NOT NULL,
token_delta BIGINT NOT NULL,
-- Fee earned by the step this entry closes, in micro-satoshis, and
-- what kind of step it was. See `db::cauldron::fees`. Nullable so
-- `ensure_fee_columns` can add them to an existing database and
-- `backfill_fees` can tell filled rows from unfilled ones.
fee_e6 INTEGER,
step_kind INTEGER
token_delta BIGINT NOT NULL
)",
)
.execute(pool)
@ -104,142 +98,6 @@ pub async fn create_table(pool: &SqlitePool) {
.unwrap();
}
/// Add the per-step fee columns to an existing database.
///
/// Safe to run on every start: `add_column_if_missing` is a no-op once they
/// exist. Deliberately adds no index — `SUM(fee_e6) ... WHERE pool = ? AND
/// effective_timestamp >= ?` is served by the existing
/// `idx_phe_pool_timestamp`, and `pool_history_entry` already carries ten
/// indexes with redundant pairs among them. Prune those before adding another.
pub async fn ensure_fee_columns(pool: &SqlitePool) {
use crate::db::cauldron::tokenlist::db_utils::add_column_if_missing;
add_column_if_missing(pool, "pool_history_entry", "fee_e6", "INTEGER")
.await
.expect("failed to add pool_history_entry.fee_e6");
add_column_if_missing(pool, "pool_history_entry", "step_kind", "INTEGER")
.await
.expect("failed to add pool_history_entry.step_kind");
}
/// Fill `fee_e6`/`step_kind` for rows written before the columns existed.
///
/// Walks one pool at a time, in `sequence` order, since a step's fee is defined
/// against the entry before it. Each pool is its own short transaction, so the
/// write lock is never held long and memory stays bounded by the largest single
/// pool rather than by the table — the alternative, reading 3.7M rows and
/// updating them in one transaction, is minutes of held lock and hundreds of
/// megabytes resident.
///
/// Resumable by construction: only rows with `fee_e6 IS NULL` are considered, so
/// an interrupted run continues where it stopped. A completed backfill costs one
/// indexed count on subsequent starts and nothing else.
///
/// Errors are logged and skipped rather than fatal. A pool whose fees cannot be
/// computed is left NULL and retried next start; taking the indexer down over a
/// display figure would be the wrong trade.
pub async fn backfill_fees(pool: &SqlitePool) {
let pending: i64 =
match sqlx::query_scalar("SELECT COUNT(*) FROM pool_history_entry WHERE fee_e6 IS NULL")
.fetch_one(pool)
.await
{
Ok(n) => n,
Err(e) => {
error!("Could not count rows needing fee backfill: {e}");
return;
}
};
if pending == 0 {
return;
}
let pools: Vec<Vec<u8>> = match sqlx::query_scalar(
"SELECT DISTINCT pool FROM pool_history_entry WHERE fee_e6 IS NULL",
)
.fetch_all(pool)
.await
{
Ok(p) => p,
Err(e) => {
error!("Could not list pools needing fee backfill: {e}");
return;
}
};
info!(
"Backfilling fees for {pending} entries across {} pools",
pools.len()
);
let started = std::time::Instant::now();
let mut written = 0i64;
for (done, pool_blob) in pools.iter().enumerate() {
match backfill_one_pool(pool, pool_blob).await {
Ok(n) => written += n,
Err(e) => error!("Fee backfill failed for one pool, will retry next start: {e}"),
}
// Give the runtime a turn: this runs before the API binds, and a tight
// loop over tens of thousands of pools would otherwise monopolise it.
if done % 500 == 499 {
info!(" ... {done}/{} pools", pools.len());
tokio::task::yield_now().await;
}
}
info!(
"Backfilled {written} pool history fees in {:.1}s",
started.elapsed().as_secs_f64()
);
}
/// One pool's chain, in its own transaction.
async fn backfill_one_pool(pool: &SqlitePool, pool_blob: &[u8]) -> Result<i64> {
let rows = sqlx::query(
"SELECT utxo, sats, token_amount
FROM pool_history_entry
WHERE pool = ?1
ORDER BY sequence ASC",
)
.bind(pool_blob)
.fetch_all(pool)
.await?;
let mut tx = pool.begin().await?;
let mut prev: Option<Reserves> = None;
let mut written = 0i64;
for row in rows {
let utxo: Vec<u8> = row.get(0);
let sats: i64 = row.get(1);
let token_amount: i64 = row.get(2);
let next = Reserves {
sats: sats.max(0) as u64,
tokens: token_amount.max(0) as u64,
};
let (fee_e6, kind) = match prev {
Some(prev) => (step_fee_e6(prev, next), classify_step(prev, next).as_i64()),
// A pool's first entry has no step behind it.
None => (0, StepKind::Unreadable.as_i64()),
};
sqlx::query("UPDATE pool_history_entry SET fee_e6 = ?, step_kind = ? WHERE utxo = ?")
.bind(fee_e6)
.bind(kind)
.bind(&utxo)
.execute(&mut *tx)
.await?;
prev = Some(next);
written += 1;
}
tx.commit().await?;
Ok(written)
}
async fn get_pool_by_utxo(
conn: &mut SqliteConnection,
utxo_hash: &OutPointHash,
@ -306,10 +164,6 @@ pub fn dummy_init_seq() {
}
}
// One more parameter than clippy's default, as in `ohlcv.rs` and
// `moria_v11/ingest.rs`. Grouping the step arguments into a struct would touch
// every call site to no benefit — they are all passing literals.
#[allow(clippy::too_many_arguments)]
pub async fn insert_pool_history_entry(
conn: &mut SqliteConnection,
pool: &OutPointHash,
@ -318,30 +172,14 @@ pub async fn insert_pool_history_entry(
first_seen_timestamp: Option<u64>,
sats_delta: i64,
token_delta: i64,
// The pool's reserves in the entry this one succeeds, or `None` for a
// pool's first entry — which earns nothing, having no step behind it.
prev: Option<Reserves>,
) -> Result<()> {
let next_seq = NEXT_SEQUENCE.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
assert!(next_seq >= 0);
// Computed here rather than by the caller so it cannot be forgotten at one
// of the two call sites, and so the fee is written in the same statement as
// the reserves it describes.
let (fee_e6, step_kind) = match (prev, cauldron.sats, cauldron.token_amount) {
(Some(prev), Some(sats), Some(token_amount)) if token_amount >= 0 => {
let next = Reserves {
sats,
tokens: token_amount as u64,
};
(step_fee_e6(prev, next), classify_step(prev, next).as_i64())
}
_ => (0, StepKind::Unreadable.as_i64()),
};
sqlx::query(
"INSERT INTO pool_history_entry (utxo, pool, token_id, txid, tx_pos, mtp_timestamp, first_seen_timestamp, sequence, sats, token_amount, sats_delta, token_delta, fee_e6, step_kind)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
"INSERT INTO pool_history_entry (utxo, pool, token_id, txid, tx_pos, mtp_timestamp, first_seen_timestamp, sequence, sats, token_amount, sats_delta, token_delta)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(utxo) DO UPDATE SET
pool = excluded.pool,
token_id = excluded.token_id,
@ -355,9 +193,7 @@ pub async fn insert_pool_history_entry(
-- history and breaking any cursor paging over it. The original value
-- is the order the state was first observed, and for a chain of
-- spends a parent is always observed before its child.
sequence = pool_history_entry.sequence,
fee_e6 = excluded.fee_e6,
step_kind = excluded.step_kind",
sequence = pool_history_entry.sequence",
)
.bind(cauldron.new_utxo_hash.expect("utxo hash on new pool history entry").to_blob())
.bind(pool.to_blob())
@ -371,8 +207,6 @@ pub async fn insert_pool_history_entry(
.bind(cauldron.token_amount)
.bind(sats_delta)
.bind(token_delta)
.bind(fee_e6)
.bind(step_kind)
.execute(&mut *conn)
.await
.map_err(|e| anyhow::anyhow!("failed to insert pool history entry. Original error: {:?}", e))?;
@ -432,7 +266,6 @@ pub async fn update_pool_history(
first_seen_timestamp,
0,
0,
None,
)
.await?;
} else {
@ -456,10 +289,6 @@ pub async fn update_pool_history(
first_seen_timestamp,
sats_delta,
token_delta,
Some(Reserves {
sats: prev_entry.sats,
tokens: prev_entry.token_amount,
}),
)
.await?;
}
@ -904,59 +733,6 @@ async fn get_pool_history_entry(
})
}
/// Summed fees per pool since `start`, for a batch of pools.
///
/// One indexed pass instead of shipping every history row to the client and
/// having it re-derive the same figure. Pools with no rows in the window are
/// absent from the result rather than reported as zero — the caller cannot
/// otherwise tell "earned nothing" from "does not exist", and on a money figure
/// those must not look alike.
pub async fn db_pool_fees(
pool: &SqlitePool,
pool_ids: &[PoolID],
start_time: u64,
) -> Result<Vec<PoolFees>> {
let mut out = Vec::with_capacity(pool_ids.len());
// One statement per pool rather than an IN list: the parameter limit is
// finite, the index is (pool, effective_timestamp) so each is a range scan,
// and it keeps the mapping back to the requested id unambiguous.
for pool_id in pool_ids {
let row = sqlx::query(
"SELECT
COALESCE(SUM(fee_e6), 0),
COALESCE(SUM(step_kind = 2), 0),
COALESCE(SUM(step_kind = 1), 0),
COALESCE(SUM(step_kind = 0), 0),
COUNT(*)
FROM pool_history_entry
WHERE pool = ?1 AND effective_timestamp >= ?2",
)
.bind(pool_id.to_blob())
.bind(start_time as i64)
.fetch_one(pool)
.await?;
let rows: i64 = row.get(4);
if rows == 0 {
continue;
}
let fee_e6: i64 = row.get(0);
out.push(PoolFees {
pool_id: pool_id.to_string(),
// A decimal string, not a float: this is money, and the consumer
// reads it into a BigNumber.
fee_sats: format!("{}.{:06}", fee_e6 / FEE_SCALE, (fee_e6 % FEE_SCALE).abs()),
trades: row.get(1),
liquidity_changes: row.get(2),
unreadable: row.get(3),
});
}
Ok(out)
}
/// Where a page of history left off.
///
/// Ordering is by `sequence`, the order states were observed. For a pool that is
@ -980,6 +756,103 @@ pub struct HistoryCursor {
pub utxo: Vec<u8>,
}
/// Summed fees per pool since `start`, for a batch of pools.
///
/// Walks each pool's reserves and values every step at the point it happened.
/// Deliberately computed rather than stored: a `fee_e6` column per row was
/// measured at only ~2x faster on the worst realistic query, which is cached
/// anyway, in exchange for 146 MB, a startup backfill, and a precision decision
/// frozen at write time. The win over the old design is doing this on the server
/// at all — clients used to download every pool's whole history to do it
/// themselves.
///
/// Pools with no rows in the window are absent from the result rather than
/// reported as zero: a caller cannot otherwise tell "earned nothing" from "does
/// not exist", and on a money figure those must not look alike.
pub async fn db_pool_fees(
pool: &SqlitePool,
pool_ids: &[PoolID],
start_time: u64,
) -> Result<Vec<PoolFees>> {
let mut out = Vec::with_capacity(pool_ids.len());
for pool_id in pool_ids {
// The row before the window seeds the walk, so the step landing on the
// window's first entry is counted. That fee was earned at the moment of
// the later entry, which is inside the window.
let seed = sqlx::query(
"SELECT sats, token_amount
FROM pool_history_entry
WHERE pool = ?1 AND effective_timestamp < ?2
ORDER BY sequence DESC
LIMIT 1",
)
.bind(pool_id.to_blob())
.bind(start_time as i64)
.fetch_optional(pool)
.await?;
let rows = sqlx::query(
"SELECT sats, token_amount
FROM pool_history_entry
WHERE pool = ?1 AND effective_timestamp >= ?2
ORDER BY sequence ASC",
)
.bind(pool_id.to_blob())
.bind(start_time as i64)
.fetch_all(pool)
.await?;
if rows.is_empty() {
continue;
}
let reserves_of = |row: &sqlx::sqlite::SqliteRow| {
let sats: i64 = row.get(0);
let tokens: i64 = row.get(1);
Reserves {
sats: sats.max(0) as u64,
tokens: tokens.max(0) as u64,
}
};
let mut fee_e6: i64 = 0;
let mut trades = 0i64;
let mut liquidity_changes = 0i64;
let mut unreadable = 0i64;
let mut prev = seed.as_ref().map(reserves_of);
for row in &rows {
let next = reserves_of(row);
match prev {
Some(prev) => {
fee_e6 = fee_e6.saturating_add(step_fee_e6(prev, next));
match classify_step(prev, next) {
StepKind::Trade => trades += 1,
StepKind::LiquidityChange => liquidity_changes += 1,
StepKind::Unreadable => unreadable += 1,
}
}
// No step behind the first entry we can see.
None => unreadable += 1,
}
prev = Some(next);
}
out.push(PoolFees {
pool_id: pool_id.to_string(),
// A decimal string, not a float: this is money, and the consumer
// reads it into a BigNumber.
fee_sats: format!("{}.{:06}", fee_e6 / FEE_SCALE, (fee_e6 % FEE_SCALE).abs()),
trades,
liquidity_changes,
unreadable,
});
}
Ok(out)
}
pub async fn db_pool_history(
pool: &SqlitePool,
pool_id: &PoolID,
@ -1212,7 +1085,6 @@ mod tests {
Some(t0 as u64),
0,
0,
None,
)
.await
.unwrap();
@ -1248,7 +1120,6 @@ mod tests {
Some(t1 as u64),
100,
-10,
None,
)
.await
.unwrap();
@ -1303,7 +1174,6 @@ mod tests {
Some(t0 as u64),
0,
0,
None,
)
.await
.unwrap();
@ -1328,10 +1198,6 @@ mod tests {
Some(t1 as u64),
(sats1 - sats0) as i64,
tokens1 - tokens0,
Some(Reserves {
sats: sats0,
tokens: tokens0 as u64,
}),
)
.await
.unwrap();
@ -1386,7 +1252,6 @@ mod tests {
None,
0,
0,
None,
)
.await
.unwrap();

2
src/db/cauldron/tokenlist/db_utils.rs
View file

@ -236,7 +236,7 @@ CREATE TABLE IF NOT EXISTS cached_token_metrics (
Ok(())
}
pub(crate) async fn add_column_if_missing(
async fn add_column_if_missing(
pool: &SqlitePool,
table: &str,
column: &str,

5
src/db/cauldron/tokenlist/test.rs
View file

@ -127,7 +127,6 @@ mod tests {
Some(t0 as u64),
0,
0,
None,
)
.await
.unwrap();
@ -164,7 +163,6 @@ mod tests {
Some(t1 as u64),
sats1 as i64,
toks1,
None,
)
.await
.unwrap();
@ -178,7 +176,6 @@ mod tests {
Some(t1 as u64),
0,
0,
None,
)
.await
.unwrap();
@ -1342,7 +1339,6 @@ mod tests {
Some(t0 as u64),
0,
0,
None,
)
.await
.unwrap();
@ -1396,7 +1392,6 @@ mod tests {
Some(t1 as u64),
0,
0,
None,
)
.await
.unwrap();

2
src/db/init.rs
View file

@ -110,8 +110,6 @@ pub async fn initialize_databases(network: &str, read_slots: ReadSlots) -> Resul
// Always-run migrations: safe on both new and existing cauldron.db
tokentoken_prepare_tables(&cauldron_db_write).await;
tokenbch_prepare_tables(&cauldron_db_write).await;
crate::db::cauldron::pool::ensure_fee_columns(&cauldron_db_write).await;
crate::db::cauldron::pool::backfill_fees(&cauldron_db_write).await;
// Initialize BCMR database
let (db_exists, bcmr_db_write, bcmr_db_read) =

7
src/db/search/test.rs
View file

@ -715,7 +715,6 @@ mod tests {
Some(thirty_days_ago as u64),
0,
0,
None,
)
.await?;
insert_pool_history_entry(
@ -726,7 +725,6 @@ mod tests {
Some(current_timestamp as u64),
2000,
1000,
None,
)
.await?;
@ -804,7 +802,6 @@ mod tests {
Some(thirty_days_ago as u64),
0,
0,
None,
)
.await?;
insert_pool_history_entry(
@ -815,7 +812,6 @@ mod tests {
Some(current_timestamp as u64),
3000,
1500,
None,
)
.await?;
@ -875,7 +871,6 @@ mod tests {
Some(thirty_days_ago as u64),
0,
0,
None,
)
.await?;
@ -933,7 +928,6 @@ mod tests {
Some(thirty_days_ago as u64),
0,
0,
None,
)
.await?;
insert_pool_history_entry(
@ -944,7 +938,6 @@ mod tests {
Some(current_timestamp as u64),
5000,
1500,
None,
)
.await?;

9
src/rpc/candlesticks/tests.rs
View file

@ -133,7 +133,6 @@ async fn setup_mock_db(pool: sqlx::SqlitePool) {
Some(TIME_1),
80_000,
2_000,
None,
)
.await
.unwrap();
@ -145,7 +144,6 @@ async fn setup_mock_db(pool: sqlx::SqlitePool) {
Some(TIME_2),
120_000,
2_000,
None,
)
.await
.unwrap();
@ -157,7 +155,6 @@ async fn setup_mock_db(pool: sqlx::SqlitePool) {
Some(TIME_3),
160_000,
2_000,
None,
)
.await
.unwrap();
@ -169,7 +166,6 @@ async fn setup_mock_db(pool: sqlx::SqlitePool) {
Some(TIME_4),
200_000,
2_000,
None,
)
.await
.unwrap();
@ -251,7 +247,6 @@ async fn insert_trade_at(
Some(ts),
deltas.0,
deltas.1,
None,
)
.await
.unwrap();
@ -431,7 +426,6 @@ async fn test_single_swap_multiple_pools() {
Some(time),
(100_000 * (i + 1)) as i64,
2_000i64,
None,
)
.await
.unwrap();
@ -534,7 +528,6 @@ async fn test_candle_prices_and_counts_volume_when_legs_cancel() {
Some(t0),
10_000,
200,
None,
)
.await
.unwrap();
@ -578,7 +571,6 @@ async fn test_candle_prices_and_counts_volume_when_legs_cancel() {
Some(t1),
10_000,
200,
None,
)
.await
.unwrap();
@ -591,7 +583,6 @@ async fn test_candle_prices_and_counts_volume_when_legs_cancel() {
Some(t1),
-10_000,
-200,
None,
)
.await
.unwrap();

6
src/rpc/pool.rs
View file

@ -189,9 +189,9 @@ const FEES_MAX_IDS: usize = 200;
/// `/cauldron/pools/fees?ids=<a,b,c>&start=<unix>`
///
/// Fees earned per pool since `start`, summed from the per-step `fee_e6` written
/// at index time. Exists so a wallet showing its total does not have to download
/// every pool's entire history and re-derive the figure in the browser.
/// Fees earned per pool since `start`. Exists so a wallet showing its total does
/// not have to download every pool's entire history and re-derive the figure in
/// the browser — which cost one large uncached request per position.
///
/// Pools absent from the response had no history in the window; a caller
/// distinguishing "earned nothing" from "no such pool" should treat absence as

6
src/rpc/price.rs
View file

@ -656,7 +656,6 @@ mod tests {
Some(1727963350),
0,
0,
None,
)
.await
.unwrap();
@ -977,7 +976,6 @@ mod tests {
Some(ts_created),
0,
0,
None,
)
.await
.unwrap();
@ -1374,7 +1372,6 @@ mod tests {
Some(TIME_1),
0,
0,
None,
)
.await
.unwrap();
@ -1582,7 +1579,7 @@ mod tests {
&OutPointHash::all_zeros(),
);
insert_new_pool(&mut conn, &ca).await.unwrap();
insert_pool_history_entry(&mut conn, &utxo_a, &ca, Some(BASE), Some(BASE), 0, 0, None)
insert_pool_history_entry(&mut conn, &utxo_a, &ca, Some(BASE), Some(BASE), 0, 0)
.await
.unwrap();
@ -1620,7 +1617,6 @@ mod tests {
Some(BASE + 200),
0,
0,
None,
)
.await
.unwrap();

3
src/rpc/tvl.rs
View file

@ -346,7 +346,6 @@ pub mod tests {
Some(1000),
0,
0,
None,
)
.await
.unwrap();
@ -358,7 +357,6 @@ pub mod tests {
Some(2000),
0,
0,
None,
)
.await
.unwrap();
@ -370,7 +368,6 @@ pub mod tests {
Some(3000),
0,
0,
None,
)
.await
.unwrap();

2
src/rpc/volume.rs
View file

@ -198,7 +198,6 @@ mod tests {
Some(current_time),
500, // sats_delta for trading activity
50, // token_delta for trading activity
None,
)
.await
.unwrap();
@ -270,7 +269,6 @@ mod tests {
Some(current_time),
500, // sats_delta for trading activity
50, // token_delta for trading activity
None,
)
.await
.unwrap();