gaspersic/solana
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Long-term ledger storage with BigTable (bp #11222)

Browse Source
This commit is contained in:
Michael Vines
2020-07-17 13:25:28 -07:00
parent 7f0d4f0656
commit 66242eab41
34 changed files with 5484 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

117
core/src/rpc.rs
View File

@@ -64,6 +64,7 @@ use std::{
str::FromStr,
sync::{Arc, RwLock},
};
use tokio::runtime;
fn new_response<T>(bank: &Bank, value: T) -> Result<RpcResponse<T>> {
let context = RpcResponseContext { slot: bank.slot() };
@@ -78,6 +79,7 @@ pub struct JsonRpcConfig {
pub identity_pubkey: Pubkey,
pub faucet_addr: Option<SocketAddr>,
pub health_check_slot_distance: u64,
pub enable_bigtable_ledger_storage: bool,
}
#[derive(Clone)]
@@ -91,6 +93,8 @@ pub struct JsonRpcRequestProcessor {
cluster_info: Arc<ClusterInfo>,
genesis_hash: Hash,
send_transaction_service: Arc<SendTransactionService>,
runtime_handle: runtime::Handle,
bigtable_ledger_storage: Option<solana_storage_bigtable::LedgerStorage>,
}
impl Metadata for JsonRpcRequestProcessor {}
@@ -145,6 +149,7 @@ impl JsonRpcRequestProcessor {
}
}
#[allow(clippy::too_many_arguments)]
pub fn new(
config: JsonRpcConfig,
bank_forks: Arc<RwLock<BankForks>>,
@@ -155,6 +160,8 @@ impl JsonRpcRequestProcessor {
cluster_info: Arc<ClusterInfo>,
genesis_hash: Hash,
send_transaction_service: Arc<SendTransactionService>,
runtime: &runtime::Runtime,
bigtable_ledger_storage: Option<solana_storage_bigtable::LedgerStorage>,
) -> Self {
Self {
config,
@@ -166,6 +173,8 @@ impl JsonRpcRequestProcessor {
cluster_info,
genesis_hash,
send_transaction_service,
runtime_handle: runtime.handle().clone(),
bigtable_ledger_storage,
}
}
@@ -507,6 +516,7 @@ impl JsonRpcRequestProcessor {
slot: Slot,
encoding: Option<UiTransactionEncoding>,
) -> Result<Option<ConfirmedBlock>> {
let encoding = encoding.unwrap_or(UiTransactionEncoding::Json);
if self.config.enable_rpc_transaction_history
&& slot
<= self
@@ -515,7 +525,15 @@ impl JsonRpcRequestProcessor {
.unwrap()
.largest_confirmed_root()
{
let result = self.blockstore.get_confirmed_block(slot, encoding);
let result = self.blockstore.get_confirmed_block(slot, Some(encoding));
if result.is_err() {
if let Some(bigtable_ledger_storage) = &self.bigtable_ledger_storage {
return Ok(self
.runtime_handle
.block_on(bigtable_ledger_storage.get_confirmed_block(slot, encoding))
.ok());
}
}
self.check_slot_cleaned_up(&result, slot)?;
Ok(result.ok())
} else {
@@ -544,9 +562,25 @@ impl JsonRpcRequestProcessor {
MAX_GET_CONFIRMED_BLOCKS_RANGE
)));
}
let lowest_slot = self.blockstore.lowest_slot();
if start_slot < lowest_slot {
// If the starting slot is lower than what's available in blockstore assume the entire
// [start_slot..end_slot] can be fetched from BigTable.
if let Some(bigtable_ledger_storage) = &self.bigtable_ledger_storage {
return Ok(self
.runtime_handle
.block_on(
bigtable_ledger_storage
.get_confirmed_blocks(start_slot, (end_slot - start_slot) as usize),
)
.unwrap_or_else(|_| vec![]));
}
}
Ok(self
.blockstore
.rooted_slot_iterator(max(start_slot, self.blockstore.lowest_slot()))
.rooted_slot_iterator(max(start_slot, lowest_slot))
.map_err(|_| Error::internal_error())?
.filter(|&slot| slot <= end_slot)
.collect())
@@ -640,6 +674,16 @@ impl JsonRpcRequestProcessor {
err,
}
})
.or_else(|| {
if let Some(bigtable_ledger_storage) = &self.bigtable_ledger_storage {
self.runtime_handle
.block_on(bigtable_ledger_storage.get_signature_status(&signature))
.map(Some)
.unwrap_or(None)
} else {
None
}
})
} else {
None
};
@@ -681,23 +725,39 @@ impl JsonRpcRequestProcessor {
&self,
signature: Signature,
encoding: Option<UiTransactionEncoding>,
) -> Result<Option<ConfirmedTransaction>> {
) -> Option<ConfirmedTransaction> {
let encoding = encoding.unwrap_or(UiTransactionEncoding::Json);
if self.config.enable_rpc_transaction_history {
Ok(self
match self
.blockstore
.get_confirmed_transaction(signature, encoding)
.get_confirmed_transaction(signature, Some(encoding))
.unwrap_or(None)
.filter(|confirmed_transaction| {
confirmed_transaction.slot
{
Some(confirmed_transaction) => {
if confirmed_transaction.slot
<= self
.block_commitment_cache
.read()
.unwrap()
.largest_confirmed_root()
}))
} else {
Ok(None)
.highest_confirmed_slot()
{
return Some(confirmed_transaction);
}
}
None => {
if let Some(bigtable_ledger_storage) = &self.bigtable_ledger_storage {
return self
.runtime_handle
.block_on(
bigtable_ledger_storage
.get_confirmed_transaction(&signature, encoding),
)
.unwrap_or(None);
}
}
}
}
None
}
pub fn get_confirmed_signatures_for_address(
@@ -707,6 +767,8 @@ impl JsonRpcRequestProcessor {
end_slot: Slot,
) -> Result<Vec<Signature>> {
if self.config.enable_rpc_transaction_history {
// TODO: Add bigtable_ledger_storage support as a part of
// https://github.com/solana-labs/solana/pull/10928
let end_slot = min(
end_slot,
self.block_commitment_cache
@@ -724,10 +786,23 @@ impl JsonRpcRequestProcessor {
}
pub fn get_first_available_block(&self) -> Result<Slot> {
Ok(self
let slot = self
.blockstore
.get_first_available_block()
.unwrap_or_default())
.unwrap_or_default();
if let Some(bigtable_ledger_storage) = &self.bigtable_ledger_storage {
let bigtable_slot = self
.runtime_handle
.block_on(bigtable_ledger_storage.get_first_available_block())
.unwrap_or(None)
.unwrap_or(slot);
if bigtable_slot < slot {
return Ok(bigtable_slot);
}
}
Ok(slot)
}
pub fn get_confirmed_signatures_for_address2(
@@ -2008,7 +2083,7 @@ impl RpcSol for RpcSolImpl {
signature_str
);
let signature = verify_signature(&signature_str)?;
meta.get_confirmed_transaction(signature, encoding)
Ok(meta.get_confirmed_transaction(signature, encoding))
}
fn get_confirmed_signatures_for_address(
@@ -2379,6 +2454,8 @@ pub mod tests {
&bank_forks,
&exit,
)),
&runtime::Runtime::new().unwrap(),
None,
);
cluster_info.insert_info(ContactInfo::new_with_pubkey_socketaddr(
@@ -2430,6 +2507,8 @@ pub mod tests {
&bank_forks,
&exit,
)),
&runtime::Runtime::new().unwrap(),
None,
);
thread::spawn(move || {
let blockhash = bank.confirmed_last_blockhash().0;
@@ -3526,6 +3605,8 @@ pub mod tests {
&bank_forks,
&exit,
)),
&runtime::Runtime::new().unwrap(),
None,
);
let req = r#"{"jsonrpc":"2.0","id":1,"method":"sendTransaction","params":["37u9WtQpcm6ULa3Vmu7ySnANv"]}"#;
@@ -3570,6 +3651,8 @@ pub mod tests {
&bank_forks,
&exit,
)),
&runtime::Runtime::new().unwrap(),
None,
);
let mut bad_transaction =
@@ -3764,6 +3847,8 @@ pub mod tests {
&bank_forks,
&exit,
)),
&runtime::Runtime::new().unwrap(),
None,
);
assert_eq!(request_processor.validator_exit(), Ok(false));
assert_eq!(exit.load(Ordering::Relaxed), false);
@@ -3796,6 +3881,8 @@ pub mod tests {
&bank_forks,
&exit,
)),
&runtime::Runtime::new().unwrap(),
None,
);
assert_eq!(request_processor.validator_exit(), Ok(true));
assert_eq!(exit.load(Ordering::Relaxed), true);
@@ -3887,6 +3974,8 @@ pub mod tests {
&bank_forks,
&exit,
)),
&runtime::Runtime::new().unwrap(),
None,
);
assert_eq!(
request_processor.get_block_commitment(0),