|
| 1 | +use super::{ForwardEvent, ForwardStage}; |
1 | 2 | use crate::chain_forward::client_state::find_headers_between; |
2 | 3 | use crate::consensus::store::{ChainStore, Nonces, StoreError}; |
| 4 | +use acto::{AcTokio, AcTokioRuntime, ActoCell, ActoInput, ActoRuntime}; |
3 | 5 | use amaru_kernel::{Hash, Header}; |
4 | | -use pallas_network::miniprotocols::chainsync::Tip; |
| 6 | +use amaru_ledger::BlockValidationResult; |
| 7 | +use gasket::messaging::tokio::ChannelRecvAdapter; |
| 8 | +use gasket::runtime::spawn_stage; |
| 9 | +use pallas_network::facades::PeerClient; |
| 10 | +use pallas_network::miniprotocols::chainsync::{NextResponse, Tip}; |
5 | 11 | use pallas_network::miniprotocols::Point; |
| 12 | +use std::future::Future; |
| 13 | +use std::sync::Arc; |
6 | 14 | use std::{collections::HashMap, fs::File, path::Path, str::FromStr}; |
| 15 | +use tokio::sync::{mpsc, Mutex}; |
| 16 | +use tracing_subscriber::EnvFilter; |
7 | 17 |
|
8 | 18 | impl ChainStore<Header> for HashMap<Hash<32>, Header> { |
9 | 19 | fn load_header(&self, hash: &Hash<32>) -> Option<Header> { |
@@ -76,6 +86,10 @@ fn point(slot: u64, hash: &str) -> Point { |
76 | 86 | Point::Specific(slot, hex(hash)) |
77 | 87 | } |
78 | 88 |
|
| 89 | +fn amaru_point(slot: u64, hash: &str) -> amaru_kernel::Point { |
| 90 | + amaru_kernel::Point::Specific(slot, hex(hash)) |
| 91 | +} |
| 92 | + |
79 | 93 | #[test] |
80 | 94 | fn test_mk_store() { |
81 | 95 | let store = mk_store(CHAIN_41); |
@@ -151,3 +165,110 @@ fn find_headers_between_tip_and_lost() { |
151 | 165 | let result = find_headers_between(&store, &tip, &points); |
152 | 166 | assert!(result.is_none(), "{result:?}"); |
153 | 167 | } |
| 168 | + |
| 169 | +#[test] |
| 170 | +fn test_chain_sync() { |
| 171 | + let _ = tracing_subscriber::fmt() |
| 172 | + .with_env_filter(EnvFilter::from_default_env()) |
| 173 | + .with_test_writer() |
| 174 | + .try_init(); |
| 175 | + |
| 176 | + // step 0a: prepare the store |
| 177 | + let store = Arc::new(Mutex::new(mk_store(CHAIN_41))); |
| 178 | + |
| 179 | + // step 0b: prepare actor to forward downstream traffic |
| 180 | + let runtime = AcTokio::new("test", 1).unwrap(); |
| 181 | + let (port_tx, mut port_rx) = mpsc::channel(1); |
| 182 | + let downstream = runtime |
| 183 | + .spawn_actor( |
| 184 | + "test", |
| 185 | + |mut cell: ActoCell<ForwardEvent, AcTokioRuntime>| async move { |
| 186 | + while let ActoInput::Message(msg) = cell.recv().await { |
| 187 | + port_tx.send(msg).await.unwrap(); |
| 188 | + } |
| 189 | + }, |
| 190 | + ) |
| 191 | + .me; |
| 192 | + |
| 193 | + // step 0c: prepare a little utility |
| 194 | + fn block_on<F: Future>(runtime: &AcTokio, f: F) -> F::Output { |
| 195 | + runtime.with_rt(|rt| rt.block_on(f)).unwrap() |
| 196 | + } |
| 197 | + |
| 198 | + // step 1: prepare the stage |
| 199 | + let (block_tx, block_rx) = mpsc::channel(1); |
| 200 | + let mut stage = ForwardStage::new(Some(downstream), store, 42, "127.0.0.1:0"); |
| 201 | + stage.upstream.connect(ChannelRecvAdapter::Mpsc(block_rx)); |
| 202 | + let tether = spawn_stage(stage, Default::default()); |
| 203 | + |
| 204 | + // step 2: wait for the listening event |
| 205 | + println!("stage state 1: {:?}", tether.check_state()); |
| 206 | + let port = block_on(&runtime, port_rx.recv()).unwrap(); |
| 207 | + let ForwardEvent::Listening(port) = port else { |
| 208 | + panic!("expected listening event, got {:?}", port); |
| 209 | + }; |
| 210 | + assert_ne!(port, 0); |
| 211 | + println!("stage state 2: {:?}", tether.check_state()); |
| 212 | + |
| 213 | + // step 3: send the block validated event to inform the stage of the current tip |
| 214 | + let span = tracing::debug_span!("whatever"); |
| 215 | + |
| 216 | + let validated = block_on(&runtime, { |
| 217 | + let block_tx = &block_tx; |
| 218 | + async move { |
| 219 | + println!("sending block validated"); |
| 220 | + block_tx |
| 221 | + .send( |
| 222 | + BlockValidationResult::BlockValidated(amaru_point(TIP_41_SLOT, TIP_41), span) |
| 223 | + .into(), |
| 224 | + ) |
| 225 | + .await |
| 226 | + .unwrap(); |
| 227 | + println!("waiting for forward event"); |
| 228 | + port_rx.recv().await.unwrap() |
| 229 | + } |
| 230 | + }); |
| 231 | + let ForwardEvent::Forward(p) = validated else { |
| 232 | + panic!("expected forward event, got {:?}", validated); |
| 233 | + }; |
| 234 | + assert_eq!(p, point(TIP_41_SLOT, TIP_41)); |
| 235 | + |
| 236 | + // step 4a: connect to the stage and prove that it is still alive |
| 237 | + println!("stage state 3: {:?}", tether.check_state()); |
| 238 | + let mut client = block_on( |
| 239 | + &runtime, |
| 240 | + PeerClient::connect(&format!("127.0.0.1:{port}"), 42), |
| 241 | + ) |
| 242 | + .unwrap(); |
| 243 | + |
| 244 | + // step 4b: find the intersection point |
| 245 | + let (p, t) = block_on(&runtime, { |
| 246 | + client |
| 247 | + .chainsync() |
| 248 | + .find_intersect(vec![point(BRANCH_41_SLOT, BRANCH_41)]) |
| 249 | + }) |
| 250 | + .unwrap(); |
| 251 | + |
| 252 | + assert_eq!(p, Some(point(BRANCH_41_SLOT, BRANCH_41))); |
| 253 | + assert_eq!(t.0, point(TIP_41_SLOT, TIP_41)); |
| 254 | + assert_eq!(t.1, TIP_41_HEIGHT); |
| 255 | + |
| 256 | + // step 5: pull headers from the stage |
| 257 | + let headers = block_on(&runtime, async move { |
| 258 | + let mut headers = Vec::new(); |
| 259 | + while let Ok(response) = client.chainsync().request_next().await { |
| 260 | + match response { |
| 261 | + NextResponse::RollForward(header, _) => headers.push(header), |
| 262 | + NextResponse::RollBackward(_, _) => panic!("unexpected roll backward"), |
| 263 | + NextResponse::Await => break, |
| 264 | + } |
| 265 | + } |
| 266 | + headers |
| 267 | + }); |
| 268 | + assert_eq!(headers.len() as u64, TIP_41_HEIGHT - BRANCH_41_HEIGHT); |
| 269 | + |
| 270 | + // prove that this is still alive - otherwise gasket will kill the stage |
| 271 | + drop(block_tx); |
| 272 | + |
| 273 | + // Note: there’s no way to shut down the gasket stage without logging to ERRORs, sorry |
| 274 | +} |
0 commit comments