diff --git a/.gitignore b/.gitignore index 56b590b9..bea08e1b 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,5 @@ target data results/* -reputation_* -revenue_* +reputation*.csv +revenue*.csv diff --git a/Cargo.lock b/Cargo.lock index 8bfc7edc..08a75393 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1178,7 +1178,7 @@ dependencies = [ "percent-encoding", "pin-project-lite", "socket2", - "system-configuration", + "system-configuration 0.6.1", "tokio", "tower-service", "tracing", @@ -1385,6 +1385,35 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" +[[package]] +name = "ldk-server-client" +version = "0.1.0" +source = "git+https://github.com/lightningdevkit/ldk-server?rev=8163f4fe139368613959bf4f10b19ee6a5b9b4ab#8163f4fe139368613959bf4f10b19ee6a5b9b4ab" +dependencies = [ + "bitcoin_hashes 0.14.0", + "hex-conservative 0.2.1", + "hyper 0.14.32", + "hyper-rustls 0.24.2", + "ldk-server-grpc", + "prost 0.11.9", + "reqwest 0.11.27", + "rustls 0.21.12", + "rustls-pemfile", +] + +[[package]] +name = "ldk-server-grpc" +version = "0.1.0" +source = "git+https://github.com/lightningdevkit/ldk-server?rev=8163f4fe139368613959bf4f10b19ee6a5b9b4ab#8163f4fe139368613959bf4f10b19ee6a5b9b4ab" +dependencies = [ + "bytes", + "http 1.3.1", + "http-body 1.0.1", + "prost 0.11.9", + "prost-build 0.11.9", + "tokio", +] + [[package]] name = "libc" version = "0.2.174" @@ -2133,6 +2162,47 @@ version = "0.8.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2b15c43186be67a4fd63bee50d0303afffcef381492ebe2c5d87f324e1b8815c" +[[package]] +name = "reqwest" +version = "0.11.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dd67538700a17451e7cba03ac727fb961abb7607553461627b97de0b89cf4a62" +dependencies = [ + "base64 0.21.7", + "bytes", + "encoding_rs", + "futures-core", + "futures-util", + "h2 0.3.26", + "http 0.2.12", + "http-body 0.4.6", + "hyper 0.14.32", + "hyper-rustls 0.24.2", + "ipnet", + "js-sys", + "log", + "mime", + "once_cell", + "percent-encoding", + "pin-project-lite", + "rustls 0.21.12", + "rustls-pemfile", + "serde", + "serde_json", + "serde_urlencoded", + "sync_wrapper 0.1.2", + "system-configuration 0.5.1", + "tokio", + "tokio-rustls 0.24.1", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", + "webpki-roots", + "winreg", +] + [[package]] name = "reqwest" version = "0.12.22" @@ -2500,7 +2570,7 @@ dependencies = [ [[package]] name = "sim-cli" version = "0.1.1" -source = "git+https://github.com/bitcoin-dev-project/sim-ln?rev=66daf6918c4d2774bf700c430a4bf4417a1377ee#66daf6918c4d2774bf700c430a4bf4417a1377ee" +source = "git+https://github.com/bitcoin-dev-project/sim-ln?rev=cefd0c4f23478fd5a9d84b49f8ce304b2c74b4e0#cefd0c4f23478fd5a9d84b49f8ce304b2c74b4e0" dependencies = [ "anyhow", "bitcoin 0.30.2", @@ -2524,7 +2594,7 @@ dependencies = [ [[package]] name = "simln-lib" version = "0.1.1" -source = "git+https://github.com/bitcoin-dev-project/sim-ln?rev=66daf6918c4d2774bf700c430a4bf4417a1377ee#66daf6918c4d2774bf700c430a4bf4417a1377ee" +source = "git+https://github.com/bitcoin-dev-project/sim-ln?rev=cefd0c4f23478fd5a9d84b49f8ce304b2c74b4e0#cefd0c4f23478fd5a9d84b49f8ce304b2c74b4e0" dependencies = [ "anyhow", "async-trait", @@ -2534,13 +2604,14 @@ dependencies = [ "expanduser", "fedimint-tonic-lnd", "hex", + "ldk-server-client", "lightning", "log", "mpsc", "rand 0.8.5", "rand_chacha 0.3.1", "rand_distr", - "reqwest", + "reqwest 0.12.22", "serde", "serde_json", "serde_millis", @@ -2678,6 +2749,17 @@ dependencies = [ "syn 2.0.104", ] +[[package]] +name = "system-configuration" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba3a3adc5c275d719af8cb4272ea1c4a6d668a777f37e115f6d11ddbc1c8e0e7" +dependencies = [ + "bitflags 1.3.2", + "core-foundation", + "system-configuration-sys 0.5.0", +] + [[package]] name = "system-configuration" version = "0.6.1" @@ -2686,7 +2768,17 @@ checksum = "3c879d448e9d986b661742763247d3693ed13609438cf3d006f51f5368a5ba6b" dependencies = [ "bitflags 2.9.1", "core-foundation", - "system-configuration-sys", + "system-configuration-sys 0.6.0", +] + +[[package]] +name = "system-configuration-sys" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a75fb188eb626b924683e3b95e3a48e63551fcfb51949de2f06a9d91dbee93c9" +dependencies = [ + "core-foundation-sys", + "libc", ] [[package]] @@ -3285,6 +3377,12 @@ dependencies = [ "untrusted 0.9.0", ] +[[package]] +name = "webpki-roots" +version = "0.25.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5f20c57d8d7db6d3b86154206ae5d8fba62dd39573114de97c2cb0578251f8e1" + [[package]] name = "which" version = "4.4.2" @@ -3575,6 +3673,16 @@ version = "0.53.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "271414315aff87387382ec3d271b52d7ae78726f5d44ac98b4f4030c91880486" +[[package]] +name = "winreg" +version = "0.50.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "524e57b2c537c0f9b1e69f1965311ec12182b4122e45035b1508cd24d2adadb1" +dependencies = [ + "cfg-if", + "windows-sys 0.48.0", +] + [[package]] name = "wit-bindgen-rt" version = "0.39.0" diff --git a/ln-simln-jamming/Cargo.toml b/ln-simln-jamming/Cargo.toml index d5ee2303..1fa55b5f 100644 --- a/ln-simln-jamming/Cargo.toml +++ b/ln-simln-jamming/Cargo.toml @@ -12,8 +12,8 @@ name = "forward-builder" path = "src/bin/forward_builder.rs" [dependencies] -simln-lib = { git = "https://github.com/bitcoin-dev-project/sim-ln", rev = "66daf6918c4d2774bf700c430a4bf4417a1377ee" } -sim-cli = { git = "https://github.com/bitcoin-dev-project/sim-ln", rev = "66daf6918c4d2774bf700c430a4bf4417a1377ee" } +simln-lib = { git = "https://github.com/bitcoin-dev-project/sim-ln", rev = "cefd0c4f23478fd5a9d84b49f8ce304b2c74b4e0" } +sim-cli = { git = "https://github.com/bitcoin-dev-project/sim-ln", rev = "cefd0c4f23478fd5a9d84b49f8ce304b2c74b4e0" } ln-resource-mgr = { path = "../ln-resource-mgr" } bitcoin = { version = "0.30.1" } async-trait = "0.1.73" diff --git a/ln-simln-jamming/src/attacks/sink.rs b/ln-simln-jamming/src/attacks/sink.rs index 51c3cd29..828fe4ff 100644 --- a/ln-simln-jamming/src/attacks/sink.rs +++ b/ln-simln-jamming/src/attacks/sink.rs @@ -337,7 +337,7 @@ mod tests { network: &[NetworkParser], ) -> SinkAttack { SinkAttack::new( - Arc::new(SimulationClock::new(1).unwrap()), + Arc::new(SimulationClock::new(std::time::SystemTime::now())), network, target, vec![attacker], diff --git a/ln-simln-jamming/src/attacks/utils.rs b/ln-simln-jamming/src/attacks/utils.rs index 91c79252..ca5f2d1f 100644 --- a/ln-simln-jamming/src/attacks/utils.rs +++ b/ln-simln-jamming/src/attacks/utils.rs @@ -431,7 +431,7 @@ mod tests { let target_channel_id: u64 = edges[2].scid.into(); - let clock = Arc::new(SimulationClock::new(1).unwrap()); + let clock = Arc::new(SimulationClock::new(std::time::SystemTime::now())); let reputation_interceptor: ReputationInterceptor = ReputationInterceptor::new_for_network(params, &edges, Arc::clone(&clock), None) .unwrap(); diff --git a/ln-simln-jamming/src/bin/forward_builder.rs b/ln-simln-jamming/src/bin/forward_builder.rs index 0b9656b0..c61d0767 100644 --- a/ln-simln-jamming/src/bin/forward_builder.rs +++ b/ln-simln-jamming/src/bin/forward_builder.rs @@ -8,18 +8,19 @@ use ln_simln_jamming::parsing::{ parse_duration, AttackType, NetworkParams, NetworkType, ReputationParams, }; use ln_simln_jamming::reputation_interceptor::{BootstrapForward, ReputationInterceptor}; -use ln_simln_jamming::{BoxError, ACCOUNTABLE_TYPE, UPGRADABLE_TYPE}; +use ln_simln_jamming::{BoxError, ACCOUNTABLE_TYPE, SIM_SEED, UPGRADABLE_TYPE}; use log::LevelFilter; use sim_cli::parsing::{create_simulation_with_network, SimParams}; use simln_lib::batched_writer::BatchedWriter; use simln_lib::clock::{Clock, SimulationClock}; use simln_lib::latency_interceptor::LatencyIntercepor; +use simln_lib::runtime::block_on_virtual_time; use simln_lib::sim_node::CustomRecords; use simln_lib::SimulationCfg; use simple_logger::SimpleLogger; use std::path::PathBuf; use std::sync::Arc; -use std::time::{Duration, UNIX_EPOCH}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::sync::Mutex; use tokio_util::task::TaskTracker; @@ -43,8 +44,7 @@ struct Cli { pub attack_type: Option, } -#[tokio::main] -async fn main() -> Result<(), BoxError> { +fn main() -> Result<(), BoxError> { SimpleLogger::new() .with_level(LevelFilter::Debug) // Lower logging from sim-ln so that we can focus on our own logs. @@ -57,13 +57,19 @@ async fn main() -> Result<(), BoxError> { let cli = Cli::parse(); + let start_time = SystemTime::now(); + block_on_virtual_time(start_time, |clock| run(clock, cli))??; + + Ok(()) +} + +async fn run(clock: Arc, cli: Cli) -> Result<(), BoxError> { let network = NetworkType::new(&cli.network, cli.attack_type, None)?; if matches!(network, NetworkType::BootstrapAttackTime(_, _, _)) { return Err("cannot run forward builder in bootstrap mode".into()); } let sim_network = network.active_network(); - let clock = Arc::new(SimulationClock::new(1000)?); let tasks = TaskTracker::new(); // Create a reputation interceptor without any bootstrap (since here we're creating the @@ -87,14 +93,14 @@ async fn main() -> Result<(), BoxError> { .to_string(), )?))), )?); - let latency_interceptor = Arc::new(LatencyIntercepor::new_poisson(300.0)?); + let latency_interceptor = Arc::new(LatencyIntercepor::new_poisson(300.0, Some(SIM_SEED))?); let sim_cfg = SimulationCfg::new( Some(cli.duration.as_secs() as u32), 3_800_000, 2.0, None, - Some(13995354354227336701), + Some(SIM_SEED), ); let exclude_pubkeys = [network.target().1] diff --git a/ln-simln-jamming/src/bin/reputation_builder.rs b/ln-simln-jamming/src/bin/reputation_builder.rs index 52fb1986..0f323b35 100644 --- a/ln-simln-jamming/src/bin/reputation_builder.rs +++ b/ln-simln-jamming/src/bin/reputation_builder.rs @@ -3,7 +3,7 @@ use std::{ fs::{File, OpenOptions}, io::Write, sync::Arc, - time::Duration, + time::{Duration, SystemTime}, }; use bitcoin::secp256k1::PublicKey; @@ -22,6 +22,7 @@ use ln_simln_jamming::{ }; use log::LevelFilter; use simln_lib::clock::SimulationClock; +use simln_lib::runtime::block_on_virtual_time; use simple_logger::SimpleLogger; #[derive(Parser)] @@ -43,8 +44,7 @@ struct Cli { pub attacker_bootstrap: Option, } -#[tokio::main] -async fn main() -> Result<(), BoxError> { +fn main() -> Result<(), BoxError> { SimpleLogger::new() .with_level(LevelFilter::Debug) .with_module_level("simln_lib::sim_node", LevelFilter::Debug) @@ -52,6 +52,14 @@ async fn main() -> Result<(), BoxError> { .unwrap(); let cli = Cli::parse(); + + let start_time = SystemTime::now(); + block_on_virtual_time(start_time, |clock| run(clock, cli))??; + + Ok(()) +} + +async fn run(clock: Arc, cli: Cli) -> Result<(), BoxError> { let forward_params: ForwardManagerParams = cli.reputation_params.into(); let network = NetworkType::new(&cli.network, cli.attack_type, cli.attacker_bootstrap)?; @@ -114,7 +122,6 @@ async fn main() -> Result<(), BoxError> { (bootstrap_records, 0) }; - let clock = Arc::new(SimulationClock::new(1)?); let reputation_clock = Arc::clone(&clock); let mut reputation_interceptor: ReputationInterceptor = ReputationInterceptor::new_for_network( diff --git a/ln-simln-jamming/src/clock.rs b/ln-simln-jamming/src/clock.rs index 45a96dfb..b90f9c7a 100644 --- a/ln-simln-jamming/src/clock.rs +++ b/ln-simln-jamming/src/clock.rs @@ -1,5 +1,4 @@ use simln_lib::clock::SimulationClock; -use std::ops::Add; use std::time::Instant; pub trait InstantClock { @@ -7,10 +6,10 @@ pub trait InstantClock { } impl InstantClock for SimulationClock { + /// Reads the current instant from tokio's clock, which tracks virtual time when the runtime is paused. The + /// returned instant is only meaningful for relative duration arithmetic, never compare it to a wall-clock + /// `Instant::now()`. fn now(&self) -> Instant { - let start_instant_std = self.get_start_instant().into(); - let elapsed = Instant::now().duration_since(start_instant_std); - - start_instant_std.add(elapsed * self.get_speedup_multiplier().into()) + tokio::time::Instant::now().into_std() } } diff --git a/ln-simln-jamming/src/lib.rs b/ln-simln-jamming/src/lib.rs index bcb9f39b..4936d92e 100644 --- a/ln-simln-jamming/src/lib.rs +++ b/ln-simln-jamming/src/lib.rs @@ -20,6 +20,9 @@ pub(crate) mod test_utils; /// Error type for errors that can be erased, includes 'static so that down-casting is possible. pub type BoxError = Box; +/// Fixes randomness used in the simulation so that runs are reproducible. +pub const SIM_SEED: u64 = 13995354354227336701; + /// The TLV type used to represent experimental accountable signals. pub const ACCOUNTABLE_TYPE: u64 = 106823; diff --git a/ln-simln-jamming/src/main.rs b/ln-simln-jamming/src/main.rs index a32496ae..763657ca 100644 --- a/ln-simln-jamming/src/main.rs +++ b/ln-simln-jamming/src/main.rs @@ -1,5 +1,6 @@ use bitcoin::secp256k1::PublicKey; use clap::Parser; +use ln_resource_mgr::forward_manager::ForwardManagerParams; use ln_simln_jamming::analysis::BatchForwardWriter; use ln_simln_jamming::attack_interceptor::AttackInterceptor; use ln_simln_jamming::attacks::AttackStatisitcs; @@ -12,13 +13,15 @@ use ln_simln_jamming::revenue_interceptor::{ PeacetimeRevenueMonitor, RevenueInterceptor, RevenueSnapshot, }; use ln_simln_jamming::{ - get_network_reputation, BoxError, NetworkReputation, ACCOUNTABLE_TYPE, UPGRADABLE_TYPE, + get_network_reputation, BoxError, NetworkReputation, ACCOUNTABLE_TYPE, SIM_SEED, + UPGRADABLE_TYPE, }; use log::LevelFilter; use sim_cli::parsing::{create_simulation_with_network, SimParams}; use simln_lib::clock::Clock; use simln_lib::clock::SimulationClock; use simln_lib::latency_interceptor::LatencyIntercepor; +use simln_lib::runtime::block_on_virtual_time; use simln_lib::sim_node::{CustomRecords, Interceptor, SimGraph, SimNode}; use simln_lib::SimulationCfg; use simple_logger::SimpleLogger; @@ -27,13 +30,19 @@ use std::fs::{self, OpenOptions}; use std::io::{BufWriter, Write}; use std::path::PathBuf; use std::sync::Arc; -use std::time::Duration; +use std::time::{Duration, SystemTime}; use tokio::select; use tokio::sync::Mutex; use tokio_util::task::TaskTracker; -#[tokio::main] -async fn main() -> Result<(), BoxError> { +/// Maximum time to run the simulation for in virutal time, used to safeguard against the clock spinning forever if an +/// attack fails to shut itself down. +const MAX_SIM_TIME_SECS: u32 = 365 * 24 * 60 * 60; + +/// The granularity in seconds with which we round our start time to be the same across runs on the same day. +const START_TIME_QUANTUM_SECS: u64 = 24 * 60 * 60; + +fn main() -> Result<(), BoxError> { let cli = Cli::parse(); let forward_params = cli.validate()?; @@ -47,6 +56,25 @@ async fn main() -> Result<(), BoxError> { .init() .unwrap(); + // We can't fix start time exactly, because LDK's graph requires a recent timestamp to validate gossip. Round to + // the nearest day so that our clock is at least fixed for runs on the same day. + let secs = SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .expect("system clock is before UNIX_EPOCH") + .as_secs(); + let start_time = + SystemTime::UNIX_EPOCH + Duration::from_secs(secs - secs % START_TIME_QUANTUM_SECS); + + block_on_virtual_time(start_time, |clock| run(clock, cli, forward_params))??; + + Ok(()) +} + +async fn run( + clock: Arc, + cli: Cli, + forward_params: ForwardManagerParams, +) -> Result<(), BoxError> { let network = NetworkType::new( &cli.network, Some(cli.attack_type.clone()), @@ -83,17 +111,16 @@ async fn main() -> Result<(), BoxError> { }) .collect(); - let clock = Arc::new(SimulationClock::new(cli.clock_speedup)?); - // Use the channel jamming interceptor and latency for simulated payments. let latency_interceptor: Arc = - Arc::new(LatencyIntercepor::new_poisson(150.0)?); + Arc::new(LatencyIntercepor::new_poisson(150.0, Some(SIM_SEED))?); let now = InstantClock::now(&*clock); - // Create a writer to store results for nodes that we care about. + // Create a writer to store results for nodes that we care about. We use real wall clock time here so that results + // don't overwrite each other. let results_dir = network - .results_dir(Clock::now(&*clock)) + .results_dir(SystemTime::now()) .ok_or("results dir none for attack")?; if !results_dir.exists() { fs::create_dir_all(&results_dir)?; @@ -254,7 +281,15 @@ async fn main() -> Result<(), BoxError> { exclude, }; - let sim_cfg = SimulationCfg::new(None, 3_800_000, 2.0, None, Some(13995354354227336701)); + // Bound the simulation at one virtual year as a safeguard. Normally the attack triggers shutdown well before + // this; the ceiling just prevents virtual time from advancing forever if an attack never terminates. + let sim_cfg = SimulationCfg::new( + Some(MAX_SIM_TIME_SECS), + 3_800_000, + 2.0, + None, + Some(SIM_SEED), + ); let (simulation, validated_activities, sim_nodes) = create_simulation_with_network( sim_cfg, &sim_params, diff --git a/ln-simln-jamming/src/parsing.rs b/ln-simln-jamming/src/parsing.rs index f07164fc..b31af349 100644 --- a/ln-simln-jamming/src/parsing.rs +++ b/ln-simln-jamming/src/parsing.rs @@ -8,7 +8,7 @@ use crate::revenue_interceptor::{PeacetimeRevenueMonitor, RevenueEvent}; use crate::BoxError; use bitcoin::secp256k1::PublicKey; use clap::{Parser, ValueEnum}; -use csv::{ReaderBuilder, StringRecord}; +use csv::StringRecord; use humantime::Duration as HumanDuration; use lightning::routing::gossip::NetworkGraph; use ln_resource_mgr::forward_manager::ForwardManagerParams; @@ -28,7 +28,7 @@ use std::str::FromStr; use std::sync::{Arc, Mutex}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::runtime::Handle; -use tokio::task::{self, JoinSet}; +use tokio::task::JoinSet; /// Default percent of good reputation pairs the target requires. pub const DEFAULT_TARGET_REP_PERCENT: &str = "50"; @@ -36,9 +36,6 @@ pub const DEFAULT_TARGET_REP_PERCENT: &str = "50"; /// Default percent of good reputation pairs with the target that the attacker requires. pub const DEFAULT_ATTACKER_REP_PERCENT: &str = "50"; -/// Default clock speedup to run with regular wall time. -pub const DEFAULT_CLOCK_SPEEDUP: &str = "1"; - /// Default htlc size that a peer must be able to get accountable to be considered as having good reputation, $10 at the /// time of writing. pub const DEFAULT_REPUTATION_MARGIN_MSAT: &str = "10000000"; @@ -401,10 +398,6 @@ pub struct Cli { #[arg(long)] pub attacker_reputation_percent: Option, - /// Speed up multiplier to add to the wall clock to run the simulation faster. - #[arg(long, default_value = DEFAULT_CLOCK_SPEEDUP)] - pub clock_speedup: u16, - /// The htlc amount that a peer must be able to get accountable to be considered as having a good reputation, expressed /// in msat. This will be converted to a fee using a base fee of 1000 msat and a proportional charge of 0.01% of the /// amount. @@ -548,7 +541,7 @@ fn network_graph( }) .collect::>(); - let clock = Arc::new(SimulationClock::new(1)?); + let clock = Arc::new(SimulationClock::new(std::time::SystemTime::now())); Ok(Arc::new( populate_network_graph(channels, clock.clone()) @@ -689,68 +682,25 @@ pub async fn history_from_file( file_path: &PathBuf, filter_duration: Option, ) -> Result, BoxError> { - let num_chunks = Handle::current().metrics().num_workers().div_ceil(2); - let breakpoints = calc_file_chunks(file_path, num_chunks as u8)?; - let file = File::open(file_path)?; - let file_size = file.metadata()?.len(); - let filter_cutoff = { - if let Some(duration) = filter_duration { - let reader = BufReader::new(file); - let mut csv_reader = csv::Reader::from_reader(reader); - let mut first_record = StringRecord::new(); - csv_reader.read_record(&mut first_record)?; - let incoming_add_ts: u64 = first_record[4].parse()?; - Some(incoming_add_ts.add(duration.as_nanos() as u64)) - } else { - None - } - }; + let mut csv_reader = csv::Reader::from_reader(BufReader::new(file)); - let mut tasks: Vec, BoxError>>> = - Vec::with_capacity(num_chunks); - - for i in 0..breakpoints.len() - 1 { - let start = breakpoints[i]; - let end = if i == num_chunks - 1 { - file_size - } else { - breakpoints[i + 1] - }; - let path_clone = file_path.clone(); - tasks.push(task::spawn(async move { - let mut file = File::open(path_clone)?; - file.seek(std::io::SeekFrom::Start(start))?; - let reader = BufReader::new(file).take(end - start); - - let mut csv_reader = if i == 0 { - csv::Reader::from_reader(reader) - } else { - ReaderBuilder::new().has_headers(false).from_reader(reader) - }; - - let mut forwards = Vec::new(); - for result in csv_reader.deserialize() { - let forward: BootstrapForward = result?; - - // If we're filtering cut off any htlc that was in flight at the cutoff point. - if let Some(cutoff) = filter_cutoff { - if forward.added_ns > cutoff || forward.settled_ns > cutoff { - break; - } - } + // The cutoff is the first forward's added_ns plus the filter duration. The file is ordered by + // timestamp, so we can stop at the first forward added or settled after the cutoff. + let mut filter_cutoff: Option = None; + let mut forwards = Vec::new(); + for result in csv_reader.deserialize() { + let forward: BootstrapForward = result?; - forwards.push(forward); + if let Some(duration) = filter_duration { + let cutoff = *filter_cutoff + .get_or_insert_with(|| forward.added_ns.add(duration.as_nanos() as u64)); + if forward.added_ns > cutoff || forward.settled_ns > cutoff { + break; } + } - Ok(forwards) - })); - } - - let mut forwards = Vec::new(); - for task in tasks { - let task_forwards = task.await??; - forwards.extend_from_slice(&task_forwards); + forwards.push(forward); } Ok(forwards) diff --git a/ln-simln-jamming/src/reputation_interceptor.rs b/ln-simln-jamming/src/reputation_interceptor.rs index 8d4412fa..2b823ba1 100644 --- a/ln-simln-jamming/src/reputation_interceptor.rs +++ b/ln-simln-jamming/src/reputation_interceptor.rs @@ -725,7 +725,7 @@ mod tests { ( ReputationInterceptor { network_nodes: Arc::new(Mutex::new(nodes)), - clock: Arc::new(SimulationClock::new(1).unwrap()), + clock: Arc::new(SimulationClock::new(std::time::SystemTime::now())), results: None, }, pubkeys, @@ -975,7 +975,7 @@ mod tests { ReputationInterceptor::new_for_network( params, &edges, - Arc::new(SimulationClock::new(1).unwrap()), + Arc::new(SimulationClock::new(std::time::SystemTime::now())), None, ) .unwrap(); @@ -1037,7 +1037,7 @@ mod tests { ReputationInterceptor::new_for_network( params, &edges, - Arc::new(SimulationClock::new(1).unwrap()), + Arc::new(SimulationClock::new(std::time::SystemTime::now())), None, ) .unwrap(); @@ -1133,7 +1133,7 @@ mod tests { async fn test_new_from_snapshot() { let (params, edges, reputation_snapshot) = setup_three_hop_network_edges(); - let clock = Arc::new(SimulationClock::new(1).unwrap()); + let clock = Arc::new(SimulationClock::new(std::time::SystemTime::now())); let interceptor: Result< ReputationInterceptor, BoxError, @@ -1206,7 +1206,7 @@ mod tests { &edges, reputation_snapshot, HashSet::new(), - Arc::new(SimulationClock::new(1).unwrap()), + Arc::new(SimulationClock::new(std::time::SystemTime::now())), None, ) .await; @@ -1234,7 +1234,7 @@ mod tests { &edges, reputation_snapshot, HashSet::new(), - Arc::new(SimulationClock::new(1).unwrap()), + Arc::new(SimulationClock::new(std::time::SystemTime::now())), None, ) .await; @@ -1263,7 +1263,7 @@ mod tests { &edges, reputation_snapshot, HashSet::new(), - Arc::new(SimulationClock::new(1).unwrap()), + Arc::new(SimulationClock::new(std::time::SystemTime::now())), None, ) .await; @@ -1290,7 +1290,7 @@ mod tests { &edges, reputation_snapshot, HashSet::new(), - Arc::new(SimulationClock::new(1).unwrap()), + Arc::new(SimulationClock::new(std::time::SystemTime::now())), None, ) .await;