use serde::de::DeserializeOwned;
use sov_rollup_interface::rpc::{
BatchIdAndOffset, BatchIdentifier, BatchResponse, EventIdentifier, ItemOrHash,
LedgerRpcProvider, QueryMode, SlotIdAndOffset, SlotIdentifier, SlotResponse, TxIdAndOffset,
TxIdentifier, TxResponse,
};
use sov_rollup_interface::stf::Event;
use tokio::sync::broadcast::Receiver;
use crate::schema::tables::{
BatchByHash, BatchByNumber, EventByNumber, SlotByHash, SlotByNumber, TxByHash, TxByNumber,
};
use crate::schema::types::{
BatchNumber, EventNumber, SlotNumber, StoredBatch, StoredSlot, TxNumber,
};
const MAX_SLOTS_PER_REQUEST: u64 = 10;
const MAX_BATCHES_PER_REQUEST: u64 = 20;
const MAX_TRANSACTIONS_PER_REQUEST: u64 = 100;
const MAX_EVENTS_PER_REQUEST: u64 = 500;
use super::LedgerDB;
impl LedgerRpcProvider for LedgerDB {
fn get_slots<B: DeserializeOwned, T: DeserializeOwned>(
&self,
slot_ids: &[sov_rollup_interface::rpc::SlotIdentifier],
query_mode: QueryMode,
) -> Result<Vec<Option<SlotResponse<B, T>>>, anyhow::Error> {
anyhow::ensure!(
slot_ids.len() <= MAX_SLOTS_PER_REQUEST as usize,
"requested too many slots. Requested: {}. Max: {}",
slot_ids.len(),
MAX_SLOTS_PER_REQUEST
);
let mut out = Vec::with_capacity(slot_ids.len());
for slot_id in slot_ids {
let slot_num = self.resolve_slot_identifier(slot_id)?;
out.push(match slot_num {
Some(num) => {
if let Some(stored_slot) = self.db.get::<SlotByNumber>(&num)? {
Some(self.populate_slot_response(num.into(), stored_slot, query_mode)?)
} else {
None
}
}
None => None,
})
}
Ok(out)
}
fn get_batches<B: DeserializeOwned, T: DeserializeOwned>(
&self,
batch_ids: &[sov_rollup_interface::rpc::BatchIdentifier],
query_mode: QueryMode,
) -> Result<Vec<Option<BatchResponse<B, T>>>, anyhow::Error> {
anyhow::ensure!(
batch_ids.len() <= MAX_BATCHES_PER_REQUEST as usize,
"requested too many batches. Requested: {}. Max: {}",
batch_ids.len(),
MAX_BATCHES_PER_REQUEST
);
let mut out = Vec::with_capacity(batch_ids.len());
for batch_id in batch_ids {
let batch_num = self.resolve_batch_identifier(batch_id)?;
out.push(match batch_num {
Some(num) => {
if let Some(stored_batch) = self.db.get::<BatchByNumber>(&num)? {
Some(self.populate_batch_response(stored_batch, query_mode)?)
} else {
None
}
}
None => None,
})
}
Ok(out)
}
fn get_transactions<T: DeserializeOwned>(
&self,
tx_ids: &[sov_rollup_interface::rpc::TxIdentifier],
_query_mode: QueryMode,
) -> Result<Vec<Option<TxResponse<T>>>, anyhow::Error> {
anyhow::ensure!(
tx_ids.len() <= MAX_TRANSACTIONS_PER_REQUEST as usize,
"requested too many transactions. Requested: {}. Max: {}",
tx_ids.len(),
MAX_TRANSACTIONS_PER_REQUEST
);
let mut out: Vec<Option<TxResponse<T>>> = Vec::with_capacity(tx_ids.len());
for id in tx_ids {
let num = self.resolve_tx_identifier(id)?;
out.push(match num {
Some(num) => {
if let Some(tx) = self.db.get::<TxByNumber>(&num)? {
Some(tx.try_into()?)
} else {
None
}
}
None => None,
})
}
Ok(out)
}
fn get_events(
&self,
event_ids: &[sov_rollup_interface::rpc::EventIdentifier],
) -> Result<Vec<Option<Event>>, anyhow::Error> {
anyhow::ensure!(
event_ids.len() <= MAX_EVENTS_PER_REQUEST as usize,
"requested too many events. Requested: {}. Max: {}",
event_ids.len(),
MAX_EVENTS_PER_REQUEST
);
let mut out = Vec::with_capacity(event_ids.len());
for id in event_ids {
let num = self.resolve_event_identifier(id)?;
out.push(match num {
Some(num) => self.db.get::<EventByNumber>(&num)?,
None => None,
})
}
Ok(out)
}
fn get_head<B: DeserializeOwned, T: DeserializeOwned>(
&self,
query_mode: QueryMode,
) -> Result<Option<SlotResponse<B, T>>, anyhow::Error> {
let next_ids = self.get_next_items_numbers();
let next_slot = next_ids.slot_number;
let head_number = next_slot.saturating_sub(1);
if let Some(stored_slot) = self
.db
.get::<SlotByNumber>(&SlotNumber(next_slot.saturating_sub(1)))?
{
return Ok(Some(self.populate_slot_response(
head_number,
stored_slot,
query_mode,
)?));
}
Ok(None)
}
fn get_slot_by_hash<B: DeserializeOwned, T: DeserializeOwned>(
&self,
hash: &[u8; 32],
query_mode: QueryMode,
) -> Result<Option<SlotResponse<B, T>>, anyhow::Error> {
self.get_slots(&[SlotIdentifier::Hash(*hash)], query_mode)
.map(|mut batches: Vec<Option<SlotResponse<B, T>>>| batches.pop().unwrap_or(None))
}
fn get_batch_by_hash<B: DeserializeOwned, T: DeserializeOwned>(
&self,
hash: &[u8; 32],
query_mode: QueryMode,
) -> Result<Option<BatchResponse<B, T>>, anyhow::Error> {
self.get_batches(&[BatchIdentifier::Hash(*hash)], query_mode)
.map(|mut batches: Vec<Option<BatchResponse<B, T>>>| batches.pop().unwrap_or(None))
}
fn get_tx_by_hash<T: DeserializeOwned>(
&self,
hash: &[u8; 32],
query_mode: QueryMode,
) -> Result<Option<TxResponse<T>>, anyhow::Error> {
self.get_transactions(&[TxIdentifier::Hash(*hash)], query_mode)
.map(|mut txs: Vec<Option<TxResponse<T>>>| txs.pop().unwrap_or(None))
}
fn get_slot_by_number<B: DeserializeOwned, T: DeserializeOwned>(
&self,
number: u64,
query_mode: QueryMode,
) -> Result<Option<SlotResponse<B, T>>, anyhow::Error> {
self.get_slots(&[SlotIdentifier::Number(number)], query_mode)
.map(|mut slots: Vec<Option<SlotResponse<B, T>>>| slots.pop().unwrap_or(None))
}
fn get_batch_by_number<B: DeserializeOwned, T: DeserializeOwned>(
&self,
number: u64,
query_mode: QueryMode,
) -> Result<Option<BatchResponse<B, T>>, anyhow::Error> {
self.get_batches(&[BatchIdentifier::Number(number)], query_mode)
.map(|mut slots| slots.pop().unwrap_or(None))
}
fn get_tx_by_number<T: DeserializeOwned>(
&self,
number: u64,
query_mode: QueryMode,
) -> Result<Option<TxResponse<T>>, anyhow::Error> {
self.get_transactions(&[TxIdentifier::Number(number)], query_mode)
.map(|mut txs| txs.pop().unwrap_or(None))
}
fn get_event_by_number(&self, number: u64) -> Result<Option<Event>, anyhow::Error> {
self.get_events(&[EventIdentifier::Number(number)])
.map(|mut events| events.pop().unwrap_or(None))
}
fn get_slots_range<B: DeserializeOwned, T: DeserializeOwned>(
&self,
start: u64,
end: u64,
query_mode: QueryMode,
) -> Result<Vec<Option<SlotResponse<B, T>>>, anyhow::Error> {
anyhow::ensure!(start <= end, "start must be <= end");
anyhow::ensure!(
end - start <= MAX_SLOTS_PER_REQUEST,
"requested slot range too large. Max: {}",
MAX_SLOTS_PER_REQUEST
);
let ids: Vec<_> = (start..=end).map(SlotIdentifier::Number).collect();
self.get_slots(&ids, query_mode)
}
fn get_batches_range<B: DeserializeOwned, T: DeserializeOwned>(
&self,
start: u64,
end: u64,
query_mode: QueryMode,
) -> Result<Vec<Option<BatchResponse<B, T>>>, anyhow::Error> {
anyhow::ensure!(start <= end, "start must be <= end");
anyhow::ensure!(
end - start <= MAX_BATCHES_PER_REQUEST,
"requested batch range too large. Max: {}",
MAX_BATCHES_PER_REQUEST
);
let ids: Vec<_> = (start..=end).map(BatchIdentifier::Number).collect();
self.get_batches(&ids, query_mode)
}
fn get_transactions_range<T: DeserializeOwned>(
&self,
start: u64,
end: u64,
query_mode: QueryMode,
) -> Result<Vec<Option<TxResponse<T>>>, anyhow::Error> {
anyhow::ensure!(start <= end, "start must be <= end");
anyhow::ensure!(
end - start <= MAX_TRANSACTIONS_PER_REQUEST,
"requested transaction range too large. Max: {}",
MAX_TRANSACTIONS_PER_REQUEST
);
let ids: Vec<_> = (start..=end).map(TxIdentifier::Number).collect();
self.get_transactions(&ids, query_mode)
}
fn subscribe_slots(&self) -> Result<Receiver<u64>, anyhow::Error> {
Ok(self.slot_subscriptions.subscribe())
}
}
impl LedgerDB {
fn resolve_slot_identifier(
&self,
slot_id: &SlotIdentifier,
) -> Result<Option<SlotNumber>, anyhow::Error> {
match slot_id {
SlotIdentifier::Hash(hash) => self.db.get::<SlotByHash>(hash),
SlotIdentifier::Number(num) => Ok(Some(SlotNumber(*num))),
}
}
fn resolve_batch_identifier(
&self,
batch_id: &BatchIdentifier,
) -> Result<Option<BatchNumber>, anyhow::Error> {
match batch_id {
BatchIdentifier::Hash(hash) => self.db.get::<BatchByHash>(hash),
BatchIdentifier::Number(num) => Ok(Some(BatchNumber(*num))),
BatchIdentifier::SlotIdAndOffset(SlotIdAndOffset { slot_id, offset }) => {
if let Some(slot_num) = self.resolve_slot_identifier(slot_id)? {
Ok(self
.db
.get::<SlotByNumber>(&slot_num)?
.map(|slot: StoredSlot| BatchNumber(slot.batches.start.0 + offset)))
} else {
Ok(None)
}
}
}
}
fn resolve_tx_identifier(
&self,
tx_id: &TxIdentifier,
) -> Result<Option<TxNumber>, anyhow::Error> {
match tx_id {
TxIdentifier::Hash(hash) => self.db.get::<TxByHash>(hash),
TxIdentifier::Number(num) => Ok(Some(TxNumber(*num))),
TxIdentifier::BatchIdAndOffset(BatchIdAndOffset { batch_id, offset }) => {
if let Some(batch_num) = self.resolve_batch_identifier(batch_id)? {
Ok(self
.db
.get::<BatchByNumber>(&batch_num)?
.map(|batch: StoredBatch| TxNumber(batch.txs.start.0 + offset)))
} else {
Ok(None)
}
}
}
}
fn resolve_event_identifier(
&self,
event_id: &EventIdentifier,
) -> Result<Option<EventNumber>, anyhow::Error> {
match event_id {
EventIdentifier::TxIdAndOffset(TxIdAndOffset { tx_id, offset }) => {
if let Some(tx_num) = self.resolve_tx_identifier(tx_id)? {
Ok(self
.db
.get::<TxByNumber>(&tx_num)?
.map(|tx| EventNumber(tx.events.start.0 + offset)))
} else {
Ok(None)
}
}
EventIdentifier::Number(num) => Ok(Some(EventNumber(*num))),
EventIdentifier::TxIdAndKey(_) => todo!(),
}
}
fn populate_slot_response<B: DeserializeOwned, T: DeserializeOwned>(
&self,
number: u64,
slot: StoredSlot,
mode: QueryMode,
) -> Result<SlotResponse<B, T>, anyhow::Error> {
Ok(match mode {
QueryMode::Compact => SlotResponse {
number,
hash: slot.hash,
batch_range: slot.batches.start.into()..slot.batches.end.into(),
batches: None,
},
QueryMode::Standard => {
let batches = self.get_batch_range(&slot.batches)?;
let batch_hashes = Some(
batches
.into_iter()
.map(|batch| ItemOrHash::Hash(batch.hash))
.collect(),
);
SlotResponse {
number,
hash: slot.hash,
batch_range: slot.batches.start.into()..slot.batches.end.into(),
batches: batch_hashes,
}
}
QueryMode::Full => {
let num_batches = (slot.batches.end.0 - slot.batches.start.0) as usize;
let mut batches = Vec::with_capacity(num_batches);
for batch in self.get_batch_range(&slot.batches)? {
batches.push(ItemOrHash::Full(self.populate_batch_response(batch, mode)?));
}
SlotResponse {
number,
hash: slot.hash,
batch_range: slot.batches.start.into()..slot.batches.end.into(),
batches: Some(batches),
}
}
})
}
fn populate_batch_response<B: DeserializeOwned, T: DeserializeOwned>(
&self,
batch: StoredBatch,
mode: QueryMode,
) -> Result<BatchResponse<B, T>, anyhow::Error> {
Ok(match mode {
QueryMode::Compact => batch.try_into()?,
QueryMode::Standard => {
let txs = self.get_tx_range(&batch.txs)?;
let tx_hashes = Some(
txs.into_iter()
.map(|tx| ItemOrHash::Hash(tx.hash))
.collect(),
);
let mut batch_response: BatchResponse<B, T> = batch.try_into()?;
batch_response.txs = tx_hashes;
batch_response
}
QueryMode::Full => {
let num_txs = (batch.txs.end.0 - batch.txs.start.0) as usize;
let mut txs = Vec::with_capacity(num_txs);
for tx in self.get_tx_range(&batch.txs)? {
txs.push(ItemOrHash::Full(tx.try_into()?));
}
let mut batch_response: BatchResponse<B, T> = batch.try_into()?;
batch_response.txs = Some(txs);
batch_response
}
})
}
}
#[cfg(test)]
mod tests {
use sov_rollup_interface::mocks::{MockBlob, MockBlock};
use sov_rollup_interface::rpc::LedgerRpcProvider;
use crate::ledger_db::{LedgerDB, SlotCommit};
#[test]
fn test_slot_subscription() {
let temp_dir = tempfile::tempdir().unwrap();
let path = temp_dir.path();
let db = LedgerDB::with_path(path).unwrap();
let mut rx = db.subscribe_slots().unwrap();
db.commit_slot(SlotCommit::<_, MockBlob, Vec<u8>>::new(MockBlock::default()))
.unwrap();
assert_eq!(rx.blocking_recv().unwrap(), 1);
}
}