storage-bigtable: Upload entries (#34099)
* Add entries table to bt init * Add entries to storage-proto * Use new Blockstore method in bigtable_upload * Add LedgerStorage::upload_confirmed_block_with_entries and use in bigtable_upload * Upload entries to bigtable
This commit is contained in:
parent
b9ef204faf
commit
573ec81fbb
|
@ -178,10 +178,10 @@ pub async fn upload_confirmed_blocks(
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
|
||||||
let _ = match blockstore.get_rooted_block(slot, true) {
|
let _ = match blockstore.get_rooted_block_with_entries(slot, true) {
|
||||||
Ok(confirmed_block) => {
|
Ok(confirmed_block_with_entries) => {
|
||||||
num_blocks_read += 1;
|
num_blocks_read += 1;
|
||||||
sender.send((slot, Some(confirmed_block)))
|
sender.send((slot, Some(confirmed_block_with_entries)))
|
||||||
}
|
}
|
||||||
Err(err) => {
|
Err(err) => {
|
||||||
warn!(
|
warn!(
|
||||||
|
@ -227,7 +227,8 @@ pub async fn upload_confirmed_blocks(
|
||||||
Some(confirmed_block) => {
|
Some(confirmed_block) => {
|
||||||
let bt = bigtable.clone();
|
let bt = bigtable.clone();
|
||||||
Some(tokio::spawn(async move {
|
Some(tokio::spawn(async move {
|
||||||
bt.upload_confirmed_block(slot, confirmed_block).await
|
bt.upload_confirmed_block_with_entries(slot, confirmed_block)
|
||||||
|
.await
|
||||||
}))
|
}))
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
|
@ -16,7 +16,7 @@ if [[ -n $BIGTABLE_EMULATOR_HOST ]]; then
|
||||||
cbt+=(-project emulator)
|
cbt+=(-project emulator)
|
||||||
fi
|
fi
|
||||||
|
|
||||||
for table in blocks tx tx-by-addr; do
|
for table in blocks entries tx tx-by-addr; do
|
||||||
(
|
(
|
||||||
set -x
|
set -x
|
||||||
"${cbt[@]}" createtable $table
|
"${cbt[@]}" createtable $table
|
||||||
|
|
|
@ -15,12 +15,13 @@ use {
|
||||||
timing::AtomicInterval,
|
timing::AtomicInterval,
|
||||||
transaction::{TransactionError, VersionedTransaction},
|
transaction::{TransactionError, VersionedTransaction},
|
||||||
},
|
},
|
||||||
solana_storage_proto::convert::{generated, tx_by_addr},
|
solana_storage_proto::convert::{entries, generated, tx_by_addr},
|
||||||
solana_transaction_status::{
|
solana_transaction_status::{
|
||||||
extract_and_fmt_memos, ConfirmedBlock, ConfirmedTransactionStatusWithSignature,
|
extract_and_fmt_memos, ConfirmedBlock, ConfirmedTransactionStatusWithSignature,
|
||||||
ConfirmedTransactionWithStatusMeta, Reward, TransactionByAddrInfo,
|
ConfirmedTransactionWithStatusMeta, Reward, TransactionByAddrInfo,
|
||||||
TransactionConfirmationStatus, TransactionStatus, TransactionStatusMeta,
|
TransactionConfirmationStatus, TransactionStatus, TransactionStatusMeta,
|
||||||
TransactionWithStatusMeta, VersionedConfirmedBlock, VersionedTransactionWithStatusMeta,
|
TransactionWithStatusMeta, VersionedConfirmedBlock, VersionedConfirmedBlockWithEntries,
|
||||||
|
VersionedTransactionWithStatusMeta,
|
||||||
},
|
},
|
||||||
std::{
|
std::{
|
||||||
collections::{HashMap, HashSet},
|
collections::{HashMap, HashSet},
|
||||||
|
@ -91,6 +92,10 @@ fn slot_to_blocks_key(slot: Slot) -> String {
|
||||||
slot_to_key(slot)
|
slot_to_key(slot)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn slot_to_entries_key(slot: Slot) -> String {
|
||||||
|
slot_to_key(slot)
|
||||||
|
}
|
||||||
|
|
||||||
fn slot_to_tx_by_addr_key(slot: Slot) -> String {
|
fn slot_to_tx_by_addr_key(slot: Slot) -> String {
|
||||||
slot_to_key(!slot)
|
slot_to_key(!slot)
|
||||||
}
|
}
|
||||||
|
@ -883,7 +888,30 @@ impl LedgerStorage {
|
||||||
"LedgerStorage::upload_confirmed_block request received: {:?}",
|
"LedgerStorage::upload_confirmed_block request received: {:?}",
|
||||||
slot
|
slot
|
||||||
);
|
);
|
||||||
|
self.upload_confirmed_block_with_entries(
|
||||||
|
slot,
|
||||||
|
VersionedConfirmedBlockWithEntries {
|
||||||
|
block: confirmed_block,
|
||||||
|
entries: vec![],
|
||||||
|
},
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn upload_confirmed_block_with_entries(
|
||||||
|
&self,
|
||||||
|
slot: Slot,
|
||||||
|
confirmed_block: VersionedConfirmedBlockWithEntries,
|
||||||
|
) -> Result<()> {
|
||||||
|
trace!(
|
||||||
|
"LedgerStorage::upload_confirmed_block_with_entries request received: {:?}",
|
||||||
|
slot
|
||||||
|
);
|
||||||
let mut by_addr: HashMap<&Pubkey, Vec<TransactionByAddrInfo>> = HashMap::new();
|
let mut by_addr: HashMap<&Pubkey, Vec<TransactionByAddrInfo>> = HashMap::new();
|
||||||
|
let VersionedConfirmedBlockWithEntries {
|
||||||
|
block: confirmed_block,
|
||||||
|
entries,
|
||||||
|
} = confirmed_block;
|
||||||
|
|
||||||
let mut tx_cells = Vec::with_capacity(confirmed_block.transactions.len());
|
let mut tx_cells = Vec::with_capacity(confirmed_block.transactions.len());
|
||||||
for (index, transaction_with_meta) in confirmed_block.transactions.iter().enumerate() {
|
for (index, transaction_with_meta) in confirmed_block.transactions.iter().enumerate() {
|
||||||
|
@ -934,6 +962,14 @@ impl LedgerStorage {
|
||||||
})
|
})
|
||||||
.collect();
|
.collect();
|
||||||
|
|
||||||
|
let num_entries = entries.len();
|
||||||
|
let entry_cell = (
|
||||||
|
slot_to_entries_key(slot),
|
||||||
|
entries::Entries {
|
||||||
|
entries: entries.into_iter().enumerate().map(Into::into).collect(),
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
let mut tasks = vec![];
|
let mut tasks = vec![];
|
||||||
|
|
||||||
if !tx_cells.is_empty() {
|
if !tx_cells.is_empty() {
|
||||||
|
@ -955,6 +991,14 @@ impl LedgerStorage {
|
||||||
}));
|
}));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if num_entries > 0 {
|
||||||
|
let conn = self.connection.clone();
|
||||||
|
tasks.push(tokio::spawn(async move {
|
||||||
|
conn.put_protobuf_cells_with_retry::<entries::Entries>("entries", &[entry_cell])
|
||||||
|
.await
|
||||||
|
}));
|
||||||
|
}
|
||||||
|
|
||||||
let mut bytes_written = 0;
|
let mut bytes_written = 0;
|
||||||
let mut maybe_first_err: Option<Error> = None;
|
let mut maybe_first_err: Option<Error> = None;
|
||||||
|
|
||||||
|
@ -995,6 +1039,7 @@ impl LedgerStorage {
|
||||||
"storage-bigtable-upload-block",
|
"storage-bigtable-upload-block",
|
||||||
("slot", slot, i64),
|
("slot", slot, i64),
|
||||||
("transactions", num_transactions, i64),
|
("transactions", num_transactions, i64),
|
||||||
|
("entries", num_entries, i64),
|
||||||
("bytes", bytes_written, i64),
|
("bytes", bytes_written, i64),
|
||||||
);
|
);
|
||||||
Ok(())
|
Ok(())
|
||||||
|
|
|
@ -6,7 +6,11 @@ fn main() -> Result<(), std::io::Error> {
|
||||||
}
|
}
|
||||||
|
|
||||||
let proto_base_path = std::path::PathBuf::from("proto");
|
let proto_base_path = std::path::PathBuf::from("proto");
|
||||||
let proto_files = ["confirmed_block.proto", "transaction_by_addr.proto"];
|
let proto_files = [
|
||||||
|
"confirmed_block.proto",
|
||||||
|
"entries.proto",
|
||||||
|
"transaction_by_addr.proto",
|
||||||
|
];
|
||||||
let mut protos = Vec::new();
|
let mut protos = Vec::new();
|
||||||
for proto_file in &proto_files {
|
for proto_file in &proto_files {
|
||||||
let proto = proto_base_path.join(proto_file);
|
let proto = proto_base_path.join(proto_file);
|
||||||
|
|
|
@ -0,0 +1,15 @@
|
||||||
|
syntax = "proto3";
|
||||||
|
|
||||||
|
package solana.storage.Entries;
|
||||||
|
|
||||||
|
message Entries {
|
||||||
|
repeated Entry entries = 1;
|
||||||
|
}
|
||||||
|
|
||||||
|
message Entry {
|
||||||
|
uint32 index = 1;
|
||||||
|
uint64 num_hashes = 2;
|
||||||
|
bytes hash = 3;
|
||||||
|
uint64 num_transactions = 4;
|
||||||
|
uint32 starting_transaction_index = 5;
|
||||||
|
}
|
|
@ -15,7 +15,7 @@ use {
|
||||||
transaction_context::TransactionReturnData,
|
transaction_context::TransactionReturnData,
|
||||||
},
|
},
|
||||||
solana_transaction_status::{
|
solana_transaction_status::{
|
||||||
ConfirmedBlock, InnerInstruction, InnerInstructions, Reward, RewardType,
|
ConfirmedBlock, EntrySummary, InnerInstruction, InnerInstructions, Reward, RewardType,
|
||||||
TransactionByAddrInfo, TransactionStatusMeta, TransactionTokenBalance,
|
TransactionByAddrInfo, TransactionStatusMeta, TransactionTokenBalance,
|
||||||
TransactionWithStatusMeta, VersionedConfirmedBlock, VersionedTransactionWithStatusMeta,
|
TransactionWithStatusMeta, VersionedConfirmedBlock, VersionedTransactionWithStatusMeta,
|
||||||
},
|
},
|
||||||
|
@ -41,6 +41,11 @@ pub mod tx_by_addr {
|
||||||
));
|
));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[allow(clippy::derive_partial_eq_without_eq)]
|
||||||
|
pub mod entries {
|
||||||
|
include!(concat!(env!("OUT_DIR"), "/solana.storage.entries.rs"));
|
||||||
|
}
|
||||||
|
|
||||||
impl From<Vec<Reward>> for generated::Rewards {
|
impl From<Vec<Reward>> for generated::Rewards {
|
||||||
fn from(rewards: Vec<Reward>) -> Self {
|
fn from(rewards: Vec<Reward>) -> Self {
|
||||||
Self {
|
Self {
|
||||||
|
@ -1189,6 +1194,18 @@ impl TryFrom<tx_by_addr::TransactionByAddr> for Vec<TransactionByAddrInfo> {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl From<(usize, EntrySummary)> for entries::Entry {
|
||||||
|
fn from((index, entry_summary): (usize, EntrySummary)) -> Self {
|
||||||
|
entries::Entry {
|
||||||
|
index: index as u32,
|
||||||
|
num_hashes: entry_summary.num_hashes,
|
||||||
|
hash: entry_summary.hash.as_ref().into(),
|
||||||
|
num_transactions: entry_summary.num_transactions,
|
||||||
|
starting_transaction_index: entry_summary.starting_transaction_index as u32,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod test {
|
mod test {
|
||||||
use {super::*, enum_iterator::all};
|
use {super::*, enum_iterator::all};
|
||||||
|
|
Loading…
Reference in New Issue