Skip to content

Commit

Permalink
fix: add metric and unbounded sender for receipt store
Browse files Browse the repository at this point in the history
Signed-off-by: Gustavo Inacio <[email protected]>
  • Loading branch information
gusinacio committed Dec 30, 2024
1 parent e018aab commit d3a0829
Show file tree
Hide file tree
Showing 2 changed files with 21 additions and 8 deletions.
9 changes: 5 additions & 4 deletions crates/service/src/tap.rs
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ use sqlx::PgPool;
use tap_core::receipt::checks::ReceiptCheck;
use thegraph_core::alloy::{primitives::Address, sol_types::Eip712Domain};
use tokio::sync::{
mpsc::{self, Sender},
mpsc::{self, UnboundedSender},
watch::Receiver,
};
use tokio_util::sync::CancellationToken;
Expand All @@ -31,7 +31,7 @@ const GRACE_PERIOD: u64 = 60;
#[derive(Clone)]
pub struct IndexerTapContext {
domain_separator: Arc<Eip712Domain>,
receipt_producer: Sender<DatabaseReceipt>,
receipt_producer: UnboundedSender<DatabaseReceipt>,
cancelation_token: CancellationToken,
}

Expand Down Expand Up @@ -60,8 +60,9 @@ impl IndexerTapContext {
}

pub async fn new(pgpool: PgPool, domain_separator: Eip712Domain) -> Self {
const MAX_RECEIPT_QUEUE_SIZE: usize = 1000;
let (tx, rx) = mpsc::channel(MAX_RECEIPT_QUEUE_SIZE);
// const MAX_RECEIPT_QUEUE_SIZE: usize = 1000;
// let (tx, rx) = mpsc::channel(MAX_RECEIPT_QUEUE_SIZE);
let (tx, rx) = mpsc::unbounded_channel();
let cancelation_token = CancellationToken::new();
let inner = InnerContext { pgpool };
Self::spawn_store_receipt_task(inner, rx, cancelation_token.clone());
Expand Down
20 changes: 16 additions & 4 deletions crates/service/src/tap/receipt_store.rs
Original file line number Diff line number Diff line change
Expand Up @@ -3,17 +3,27 @@

use anyhow::anyhow;
use bigdecimal::num_bigint::BigInt;
use lazy_static::lazy_static;
use prometheus::{register_gauge, Gauge};
use sqlx::{types::BigDecimal, PgPool};
use tap_core::{
manager::adapters::ReceiptStore,
receipt::{state::Checking, ReceiptWithState},
};
use thegraph_core::alloy::{hex::ToHexExt, sol_types::Eip712Domain};
use tokio::{sync::mpsc::Receiver, task::JoinHandle};
use tokio::{sync::mpsc::UnboundedReceiver, task::JoinHandle};
use tokio_util::sync::CancellationToken;

use super::{AdapterError, IndexerTapContext};

lazy_static! {
pub static ref RECEIPTS_IN_QUEUE: Gauge = register_gauge!(
"indexer_receipts_in_queue_total",
"Total receipts waiting to be stored",
)
.unwrap();
}

#[derive(Clone)]
pub struct InnerContext {
pub pgpool: PgPool,
Expand Down Expand Up @@ -74,7 +84,7 @@ impl InnerContext {
impl IndexerTapContext {
pub fn spawn_store_receipt_task(
inner_context: InnerContext,
mut receiver: Receiver<DatabaseReceipt>,
mut receiver: UnboundedReceiver<DatabaseReceipt>,
cancelation_token: CancellationToken,
) -> JoinHandle<()> {
const BUFFER_SIZE: usize = 100;
Expand All @@ -83,7 +93,8 @@ impl IndexerTapContext {
let mut buffer = Vec::with_capacity(BUFFER_SIZE);
tokio::select! {
biased;
_ = receiver.recv_many(&mut buffer, BUFFER_SIZE) => {
size = receiver.recv_many(&mut buffer, BUFFER_SIZE) => {
RECEIPTS_IN_QUEUE.sub(size as f64);
if let Err(e) = inner_context.store_receipts(buffer).await {
tracing::error!("Failed to store receipts: {}", e);
}
Expand All @@ -104,10 +115,11 @@ impl ReceiptStore for IndexerTapContext {
receipt: ReceiptWithState<Checking>,
) -> Result<u64, Self::AdapterError> {
let db_receipt = DatabaseReceipt::from_receipt(receipt, &self.domain_separator)?;
self.receipt_producer.send(db_receipt).await.map_err(|e| {
self.receipt_producer.send(db_receipt).map_err(|e| {
tracing::error!("Failed to queue receipt for storage: {}", e);
anyhow!(e)
})?;
RECEIPTS_IN_QUEUE.inc();

// We don't need receipt_ids
Ok(0)
Expand Down

0 comments on commit d3a0829

Please sign in to comment.