2019-02-21 15:16:09 -08:00
|
|
|
//! The `blockstream_service` implements optional streaming of entries and block metadata
|
|
|
|
//! using the `blockstream` module, providing client services such as a block explorer with
|
2019-02-11 10:36:59 -08:00
|
|
|
//! real-time access to entries.
|
|
|
|
|
2019-02-28 01:14:37 -08:00
|
|
|
use crate::blockstream::BlockstreamEvents;
|
2019-02-11 10:36:59 -08:00
|
|
|
#[cfg(test)]
|
2019-02-21 15:16:09 -08:00
|
|
|
use crate::blockstream::MockBlockstream as Blockstream;
|
2019-02-12 15:59:02 -08:00
|
|
|
#[cfg(not(test))]
|
2019-02-21 15:16:09 -08:00
|
|
|
use crate::blockstream::SocketBlockstream as Blockstream;
|
2019-02-28 01:14:37 -08:00
|
|
|
use crate::blocktree::Blocktree;
|
2019-02-11 10:36:59 -08:00
|
|
|
use crate::result::{Error, Result};
|
|
|
|
use crate::service::Service;
|
2019-02-28 01:14:37 -08:00
|
|
|
use solana_sdk::pubkey::Pubkey;
|
2019-02-11 10:36:59 -08:00
|
|
|
use std::sync::atomic::{AtomicBool, Ordering};
|
2019-02-28 01:14:37 -08:00
|
|
|
use std::sync::mpsc::{Receiver, RecvTimeoutError};
|
2019-02-25 14:33:42 -08:00
|
|
|
use std::sync::Arc;
|
2019-02-11 10:36:59 -08:00
|
|
|
use std::thread::{self, Builder, JoinHandle};
|
|
|
|
use std::time::Duration;
|
|
|
|
|
2019-02-21 15:16:09 -08:00
|
|
|
pub struct BlockstreamService {
|
|
|
|
t_blockstream: JoinHandle<()>,
|
2019-02-11 10:36:59 -08:00
|
|
|
}
|
|
|
|
|
2019-02-21 15:16:09 -08:00
|
|
|
impl BlockstreamService {
|
2019-02-11 10:36:59 -08:00
|
|
|
#[allow(clippy::new_ret_no_self)]
|
|
|
|
pub fn new(
|
2019-02-28 01:14:37 -08:00
|
|
|
slot_full_receiver: Receiver<(u64, Pubkey)>,
|
|
|
|
blocktree: Arc<Blocktree>,
|
2019-02-21 15:16:09 -08:00
|
|
|
blockstream_socket: String,
|
2019-03-04 20:50:02 -08:00
|
|
|
exit: &Arc<AtomicBool>,
|
2019-02-28 01:14:37 -08:00
|
|
|
) -> Self {
|
2019-02-25 14:33:42 -08:00
|
|
|
let mut blockstream = Blockstream::new(blockstream_socket);
|
2019-03-04 20:50:02 -08:00
|
|
|
let exit = exit.clone();
|
2019-02-21 15:16:09 -08:00
|
|
|
let t_blockstream = Builder::new()
|
|
|
|
.name("solana-blockstream".to_string())
|
2019-02-11 10:36:59 -08:00
|
|
|
.spawn(move || loop {
|
|
|
|
if exit.load(Ordering::Relaxed) {
|
|
|
|
break;
|
|
|
|
}
|
2019-02-28 01:14:37 -08:00
|
|
|
if let Err(e) =
|
|
|
|
Self::process_entries(&slot_full_receiver, &blocktree, &mut blockstream)
|
|
|
|
{
|
2019-02-11 10:36:59 -08:00
|
|
|
match e {
|
|
|
|
Error::RecvTimeoutError(RecvTimeoutError::Disconnected) => break,
|
|
|
|
Error::RecvTimeoutError(RecvTimeoutError::Timeout) => (),
|
|
|
|
_ => info!("Error from process_entries: {:?}", e),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
})
|
|
|
|
.unwrap();
|
2019-02-28 01:14:37 -08:00
|
|
|
Self { t_blockstream }
|
2019-02-11 10:36:59 -08:00
|
|
|
}
|
|
|
|
fn process_entries(
|
2019-02-28 01:14:37 -08:00
|
|
|
slot_full_receiver: &Receiver<(u64, Pubkey)>,
|
|
|
|
blocktree: &Arc<Blocktree>,
|
2019-02-21 15:16:09 -08:00
|
|
|
blockstream: &mut Blockstream,
|
2019-02-11 10:36:59 -08:00
|
|
|
) -> Result<()> {
|
|
|
|
let timeout = Duration::new(1, 0);
|
2019-02-28 01:14:37 -08:00
|
|
|
let (slot, slot_leader) = slot_full_receiver.recv_timeout(timeout)?;
|
|
|
|
|
|
|
|
let entries = blocktree.get_slot_entries(slot, 0, None).unwrap();
|
|
|
|
let blocktree_meta = blocktree.meta(slot).unwrap().unwrap();
|
|
|
|
let _parent_slot = if slot == 0 {
|
|
|
|
None
|
|
|
|
} else {
|
|
|
|
Some(blocktree_meta.parent_slot)
|
|
|
|
};
|
|
|
|
let ticks_per_slot = entries
|
|
|
|
.iter()
|
|
|
|
.filter(|entry| entry.is_tick())
|
|
|
|
.fold(0, |acc, _| acc + 1);
|
|
|
|
let mut tick_height = if slot > 0 {
|
|
|
|
ticks_per_slot * slot - 1
|
|
|
|
} else {
|
|
|
|
0
|
|
|
|
};
|
|
|
|
|
|
|
|
for (i, entry) in entries.iter().enumerate() {
|
|
|
|
if entry.is_tick() {
|
|
|
|
tick_height += 1;
|
2019-02-12 22:45:00 -08:00
|
|
|
}
|
2019-02-21 15:16:09 -08:00
|
|
|
blockstream
|
2019-03-09 19:28:43 -08:00
|
|
|
.emit_entry_event(slot, tick_height, &slot_leader, &entry)
|
2019-02-13 10:42:56 -08:00
|
|
|
.unwrap_or_else(|e| {
|
2019-02-21 15:16:09 -08:00
|
|
|
debug!("Blockstream error: {:?}, {:?}", e, blockstream.output);
|
2019-02-13 10:42:56 -08:00
|
|
|
});
|
2019-02-28 01:14:37 -08:00
|
|
|
if i == entries.len() - 1 {
|
|
|
|
blockstream
|
2019-03-09 19:28:43 -08:00
|
|
|
.emit_block_event(slot, tick_height, &slot_leader, entry.hash)
|
2019-02-28 01:14:37 -08:00
|
|
|
.unwrap_or_else(|e| {
|
|
|
|
debug!("Blockstream error: {:?}, {:?}", e, blockstream.output);
|
|
|
|
});
|
2019-02-12 22:45:00 -08:00
|
|
|
}
|
|
|
|
}
|
2019-02-11 10:36:59 -08:00
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-02-21 15:16:09 -08:00
|
|
|
impl Service for BlockstreamService {
|
2019-02-11 10:36:59 -08:00
|
|
|
type JoinReturnType = ();
|
|
|
|
|
|
|
|
fn join(self) -> thread::Result<()> {
|
2019-02-21 15:16:09 -08:00
|
|
|
self.t_blockstream.join()
|
2019-02-11 10:36:59 -08:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
#[cfg(test)]
|
|
|
|
mod test {
|
|
|
|
use super::*;
|
2019-02-28 07:46:42 -08:00
|
|
|
use crate::blocktree::create_new_tmp_ledger;
|
2019-02-28 01:14:37 -08:00
|
|
|
use crate::entry::{create_ticks, Entry};
|
2019-03-28 23:24:09 -07:00
|
|
|
use bincode::deserialize;
|
2019-02-11 10:36:59 -08:00
|
|
|
use chrono::{DateTime, FixedOffset};
|
|
|
|
use serde_json::Value;
|
2019-02-28 01:14:37 -08:00
|
|
|
use solana_sdk::genesis_block::GenesisBlock;
|
2019-02-11 10:36:59 -08:00
|
|
|
use solana_sdk::hash::Hash;
|
2019-02-12 22:45:00 -08:00
|
|
|
use solana_sdk::signature::{Keypair, KeypairUtil};
|
|
|
|
use solana_sdk::system_transaction::SystemTransaction;
|
2019-02-28 01:14:37 -08:00
|
|
|
use std::sync::mpsc::channel;
|
2019-02-11 10:36:59 -08:00
|
|
|
|
|
|
|
#[test]
|
2019-02-25 14:33:42 -08:00
|
|
|
fn test_blockstream_service_process_entries() {
|
2019-02-12 22:45:00 -08:00
|
|
|
let ticks_per_slot = 5;
|
2019-02-25 14:33:42 -08:00
|
|
|
let leader_id = Keypair::new().pubkey();
|
2019-02-16 08:23:29 -08:00
|
|
|
|
2019-02-28 01:14:37 -08:00
|
|
|
// Set up genesis block and blocktree
|
|
|
|
let (mut genesis_block, _mint_keypair) = GenesisBlock::new(1000);
|
|
|
|
genesis_block.ticks_per_slot = ticks_per_slot;
|
2019-02-28 07:46:42 -08:00
|
|
|
|
2019-03-02 10:25:16 -08:00
|
|
|
let (ledger_path, _blockhash) = create_new_tmp_ledger!(&genesis_block);
|
2019-03-14 15:18:37 -07:00
|
|
|
let blocktree = Blocktree::open(&ledger_path).unwrap();
|
2019-02-28 01:14:37 -08:00
|
|
|
|
2019-02-21 15:16:09 -08:00
|
|
|
// Set up blockstream
|
2019-02-25 14:33:42 -08:00
|
|
|
let mut blockstream = Blockstream::new("test_stream".to_string());
|
2019-02-11 10:36:59 -08:00
|
|
|
|
2019-02-28 01:14:37 -08:00
|
|
|
// Set up dummy channel to receive a full-slot notification
|
|
|
|
let (slot_full_sender, slot_full_receiver) = channel();
|
|
|
|
|
|
|
|
// Create entries - 4 ticks + 1 populated entry + 1 tick
|
|
|
|
let mut entries = create_ticks(4, Hash::default());
|
|
|
|
|
2019-02-12 22:45:00 -08:00
|
|
|
let keypair = Keypair::new();
|
2019-03-02 10:25:16 -08:00
|
|
|
let mut blockhash = entries[3].hash;
|
2019-03-09 19:28:43 -08:00
|
|
|
let tx = SystemTransaction::new_account(&keypair, &keypair.pubkey(), 1, Hash::default(), 0);
|
2019-03-02 10:25:16 -08:00
|
|
|
let entry = Entry::new(&mut blockhash, 1, vec![tx]);
|
|
|
|
blockhash = entry.hash;
|
2019-02-28 01:14:37 -08:00
|
|
|
entries.push(entry);
|
2019-03-02 10:25:16 -08:00
|
|
|
let final_tick = create_ticks(1, blockhash);
|
2019-02-28 01:14:37 -08:00
|
|
|
entries.extend_from_slice(&final_tick);
|
|
|
|
|
|
|
|
let expected_entries = entries.clone();
|
|
|
|
let expected_tick_heights = [5, 6, 7, 8, 8, 9];
|
|
|
|
|
2019-03-14 15:18:37 -07:00
|
|
|
blocktree
|
|
|
|
.write_entries(1, 0, 0, ticks_per_slot, &entries)
|
|
|
|
.unwrap();
|
2019-02-28 01:14:37 -08:00
|
|
|
|
|
|
|
slot_full_sender.send((1, leader_id)).unwrap();
|
2019-02-21 15:16:09 -08:00
|
|
|
BlockstreamService::process_entries(
|
2019-02-28 01:14:37 -08:00
|
|
|
&slot_full_receiver,
|
|
|
|
&Arc::new(blocktree),
|
2019-02-21 15:16:09 -08:00
|
|
|
&mut blockstream,
|
2019-02-11 10:36:59 -08:00
|
|
|
)
|
|
|
|
.unwrap();
|
2019-02-28 01:14:37 -08:00
|
|
|
assert_eq!(blockstream.entries().len(), 7);
|
2019-02-11 10:36:59 -08:00
|
|
|
|
2019-02-21 15:16:09 -08:00
|
|
|
let (entry_events, block_events): (Vec<Value>, Vec<Value>) = blockstream
|
2019-02-12 22:45:00 -08:00
|
|
|
.entries()
|
|
|
|
.iter()
|
|
|
|
.map(|item| {
|
|
|
|
let json: Value = serde_json::from_str(&item).unwrap();
|
|
|
|
let dt_str = json["dt"].as_str().unwrap();
|
|
|
|
// Ensure `ts` field parses as valid DateTime
|
|
|
|
let _dt: DateTime<FixedOffset> = DateTime::parse_from_rfc3339(dt_str).unwrap();
|
|
|
|
json
|
|
|
|
})
|
|
|
|
.partition(|json| {
|
|
|
|
let item_type = json["t"].as_str().unwrap();
|
|
|
|
item_type == "entry"
|
|
|
|
});
|
|
|
|
for (i, json) in entry_events.iter().enumerate() {
|
2019-02-20 10:08:37 -08:00
|
|
|
let height = json["h"].as_u64().unwrap();
|
|
|
|
assert_eq!(height, expected_tick_heights[i]);
|
2019-02-11 10:36:59 -08:00
|
|
|
let entry_obj = json["entry"].clone();
|
2019-02-12 22:45:00 -08:00
|
|
|
let tx = entry_obj["transactions"].as_array().unwrap();
|
2019-03-28 23:24:09 -07:00
|
|
|
let entry: Entry;
|
2019-02-12 22:45:00 -08:00
|
|
|
if tx.len() == 0 {
|
2019-03-28 23:24:09 -07:00
|
|
|
entry = serde_json::from_value(entry_obj).unwrap();
|
|
|
|
} else {
|
|
|
|
let entry_json = entry_obj.as_object().unwrap();
|
|
|
|
entry = Entry {
|
|
|
|
num_hashes: entry_json.get("num_hashes").unwrap().as_u64().unwrap(),
|
|
|
|
hash: serde_json::from_value(entry_json.get("hash").unwrap().clone()).unwrap(),
|
|
|
|
transactions: entry_json
|
|
|
|
.get("transactions")
|
|
|
|
.unwrap()
|
|
|
|
.as_array()
|
|
|
|
.unwrap()
|
|
|
|
.into_iter()
|
|
|
|
.map(|tx| {
|
|
|
|
let tx_vec: Vec<u8> = serde_json::from_value(tx.clone()).unwrap();
|
|
|
|
deserialize(&tx_vec).unwrap()
|
|
|
|
})
|
|
|
|
.collect(),
|
|
|
|
};
|
2019-02-12 22:45:00 -08:00
|
|
|
}
|
2019-03-28 23:24:09 -07:00
|
|
|
assert_eq!(entry, expected_entries[i]);
|
2019-02-11 10:36:59 -08:00
|
|
|
}
|
2019-02-12 22:45:00 -08:00
|
|
|
for json in block_events {
|
2019-02-14 14:03:37 -08:00
|
|
|
let slot = json["s"].as_u64().unwrap();
|
2019-02-28 01:14:37 -08:00
|
|
|
assert_eq!(1, slot);
|
2019-02-12 22:45:00 -08:00
|
|
|
let height = json["h"].as_u64().unwrap();
|
2019-02-28 01:14:37 -08:00
|
|
|
assert_eq!(2 * ticks_per_slot - 1, height);
|
2019-02-12 22:45:00 -08:00
|
|
|
}
|
2019-02-11 10:36:59 -08:00
|
|
|
}
|
|
|
|
}
|