1mod config;
2
3use std::collections::{BTreeMap, HashMap, HashSet};
4use std::ops::ControlFlow;
5use std::path::PathBuf;
6use std::str::FromStr;
7use std::time::Duration;
8use std::{env, fs, iter};
9
10use anyhow::{Context, Result, anyhow, bail};
11use bitcoincore_rpc::bitcoin::Network;
12use fedimint_api_client::api::DynGlobalApi;
13use fedimint_api_client::api::net::Connector;
14use fedimint_client_module::module::ClientModule;
15use fedimint_core::admin_client::{ServerStatusLegacy, SetupStatus};
16use fedimint_core::config::{ClientConfig, ServerModuleConfigGenParamsRegistry, load_from_file};
17use fedimint_core::core::LEGACY_HARDCODED_INSTANCE_ID_WALLET;
18use fedimint_core::envs::BitcoinRpcConfig;
19use fedimint_core::invite_code::InviteCode;
20use fedimint_core::module::registry::ModuleDecoderRegistry;
21use fedimint_core::module::{ApiAuth, ModuleCommon};
22use fedimint_core::runtime::block_in_place;
23use fedimint_core::task::block_on;
24use fedimint_core::task::jit::JitTryAnyhow;
25use fedimint_core::util::SafeUrl;
26use fedimint_core::{Amount, NumPeers, PeerId};
27use fedimint_gateway_common::WithdrawResponse;
28use fedimint_logging::LOG_DEVIMINT;
29use fedimint_server::config::ConfigGenParams;
30use fedimint_testing_core::config::local_config_gen_params;
31use fedimint_testing_core::node_type::LightningNodeType;
32use fedimint_wallet_client::WalletClientModule;
33use fedimint_wallet_client::config::WalletClientConfig;
34use fs_lock::FileLock;
35use futures::future::{join_all, try_join_all};
36use rand::Rng;
37use tokio::task::{JoinSet, spawn_blocking};
38use tokio::time::Instant;
39use tracing::{debug, info};
40
41use super::external::Bitcoind;
42use super::util::{Command, ProcessHandle, ProcessManager, cmd};
43use super::vars::utf8;
44use crate::envs::{FM_CLIENT_DIR_ENV, FM_DATA_DIR_ENV};
45use crate::util::{FedimintdCmd, poll, poll_simple, poll_with_timeout};
46use crate::version_constants::VERSION_0_7_0_ALPHA;
47use crate::{poll_eq, vars};
48
49pub const PORTS_PER_FEDIMINTD: u16 = 4;
52pub const FEDIMINTD_P2P_PORT_OFFSET: u16 = 0;
54pub const FEDIMINTD_API_PORT_OFFSET: u16 = 1;
56pub const FEDIMINTD_UI_PORT_OFFSET: u16 = 2;
58pub const FEDIMINTD_METRICS_PORT_OFFSET: u16 = 3;
60
61#[derive(Clone)]
62pub struct Federation {
63 pub members: BTreeMap<usize, Fedimintd>,
65 pub vars: BTreeMap<usize, vars::Fedimintd>,
66 pub bitcoind: Bitcoind,
67
68 client: JitTryAnyhow<Client>,
70}
71
72impl Drop for Federation {
73 fn drop(&mut self) {
74 block_in_place(|| {
75 block_on(async {
76 let mut set = JoinSet::new();
77
78 while let Some((_id, fedimintd)) = self.members.pop_first() {
79 set.spawn(async { drop(fedimintd) });
80 }
81 while (set.join_next().await).is_some() {}
82 });
83 });
84 }
85}
86#[derive(Clone)]
88pub struct Client {
89 name: String,
90}
91
92impl Client {
93 fn clients_dir() -> PathBuf {
94 let data_dir: PathBuf = env::var(FM_DATA_DIR_ENV)
95 .expect("FM_DATA_DIR_ENV not set")
96 .parse()
97 .expect("FM_DATA_DIR_ENV invalid");
98 data_dir.join("clients")
99 }
100
101 fn client_dir(&self) -> PathBuf {
102 Self::clients_dir().join(&self.name)
103 }
104
105 pub fn client_name_lock(name: &str) -> Result<FileLock> {
106 let lock_path = Self::clients_dir().join(format!(".{name}.lock"));
107 let file_lock = std::fs::OpenOptions::new()
108 .write(true)
109 .create(true)
110 .truncate(true)
111 .open(&lock_path)
112 .with_context(|| format!("Failed to open {}", lock_path.display()))?;
113
114 fs_lock::FileLock::new_exclusive(file_lock)
115 .with_context(|| format!("Failed to lock {}", lock_path.display()))
116 }
117
118 pub async fn create(name: impl ToString) -> Result<Client> {
120 let name = name.to_string();
121 spawn_blocking(move || {
122 let _lock = Self::client_name_lock(&name);
123 for i in 0u64.. {
124 let client = Self {
125 name: format!("{name}-{i}"),
126 };
127
128 if !client.client_dir().exists() {
129 std::fs::create_dir_all(client.client_dir())?;
130 return Ok(client);
131 }
132 }
133 unreachable!()
134 })
135 .await?
136 }
137
138 pub fn open_or_create(name: &str) -> Result<Client> {
140 block_in_place(|| {
141 let _lock = Self::client_name_lock(name);
142 let client = Self {
143 name: format!("{name}-0"),
144 };
145 if !client.client_dir().exists() {
146 std::fs::create_dir_all(client.client_dir())?;
147 }
148 Ok(client)
149 })
150 }
151
152 pub async fn join_federation(&self, invite_code: String) -> Result<()> {
154 debug!(target: LOG_DEVIMINT, "Joining federation with the main client");
155 cmd!(self, "join-federation", invite_code).run().await?;
156
157 Ok(())
158 }
159
160 pub async fn restore_federation(&self, invite_code: String, mnemonic: String) -> Result<()> {
162 debug!(target: LOG_DEVIMINT, "Joining federation with restore procedure");
163 cmd!(
164 self,
165 "restore",
166 "--invite-code",
167 invite_code,
168 "--mnemonic",
169 mnemonic
170 )
171 .run()
172 .await?;
173
174 Ok(())
175 }
176
177 pub async fn new_restored(&self, name: &str, invite_code: String) -> Result<Self> {
179 let restored = Self::open_or_create(name)?;
180
181 let mnemonic = cmd!(self, "print-secret").out_json().await?["secret"]
182 .as_str()
183 .unwrap()
184 .to_owned();
185
186 debug!(target: LOG_DEVIMINT, name, "Restoring from mnemonic");
187 cmd!(
188 restored,
189 "restore",
190 "--invite-code",
191 invite_code,
192 "--mnemonic",
193 mnemonic
194 )
195 .run()
196 .await?;
197
198 Ok(restored)
199 }
200
201 pub async fn new_forked(&self, name: impl ToString) -> Result<Client> {
204 let new = Client::create(name).await?;
205
206 cmd!(
207 "cp",
208 "-R",
209 self.client_dir().join("client.db").display(),
210 new.client_dir().display()
211 )
212 .run()
213 .await?;
214
215 Ok(new)
216 }
217
218 pub async fn balance(&self) -> Result<u64> {
219 Ok(cmd!(self, "info").out_json().await?["total_amount_msat"]
220 .as_u64()
221 .unwrap())
222 }
223
224 pub async fn get_deposit_addr(&self) -> Result<(String, String)> {
225 let deposit = cmd!(self, "deposit-address").out_json().await?;
226 Ok((
227 deposit["address"].as_str().unwrap().to_string(),
228 deposit["operation_id"].as_str().unwrap().to_string(),
229 ))
230 }
231
232 pub async fn await_deposit(&self, operation_id: &str) -> Result<()> {
233 cmd!(self, "await-deposit", operation_id).run().await
234 }
235
236 pub fn cmd(&self) -> Command {
237 cmd!(
238 crate::util::get_fedimint_cli_path(),
239 format!("--data-dir={}", self.client_dir().display())
240 )
241 }
242
243 pub fn get_name(&self) -> &str {
244 &self.name
245 }
246
247 pub async fn get_session_count(&self) -> Result<u64> {
249 cmd!(self, "dev", "session-count").out_json().await?["count"]
250 .as_u64()
251 .context("count field wasn't a number")
252 }
253
254 pub async fn wait_complete(&self) -> Result<()> {
256 cmd!(self, "dev", "wait-complete").run().await
257 }
258
259 pub async fn wait_session(&self) -> anyhow::Result<()> {
261 info!("Waiting for a new session");
262 let session_count = self.get_session_count().await?;
263 self.wait_session_outcome(session_count).await?;
264 Ok(())
265 }
266
267 pub async fn wait_session_outcome(&self, session_count: u64) -> anyhow::Result<()> {
269 let timeout = {
270 let current_session_count = self.get_session_count().await?;
271 let sessions_to_wait = session_count.saturating_sub(current_session_count) + 1;
272 let session_duration_seconds = 180;
273 Duration::from_secs(sessions_to_wait * session_duration_seconds)
274 };
275
276 let start = Instant::now();
277 poll_with_timeout("Waiting for a new session", timeout, || async {
278 info!("Awaiting session outcome {session_count}");
279 match cmd!(self, "dev", "api", "await_session_outcome", session_count)
280 .run()
281 .await
282 {
283 Err(e) => Err(ControlFlow::Continue(e)),
284 Ok(()) => Ok(()),
285 }
286 })
287 .await?;
288
289 let session_found_in = start.elapsed();
290 info!("session found in {session_found_in:?}");
291 Ok(())
292 }
293}
294
295impl Federation {
296 pub async fn new(
297 process_mgr: &ProcessManager,
298 bitcoind: Bitcoind,
299 skip_setup: bool,
300 pre_dkg: bool,
301 fed_index: usize,
303 federation_name: String,
304 ) -> Result<Self> {
305 let num_peers = NumPeers::from(process_mgr.globals.FM_FED_SIZE);
306 let mut members = BTreeMap::new();
307 let mut peer_to_env_vars_map = BTreeMap::new();
308
309 let peers: Vec<_> = num_peers.peer_ids().collect();
310 let params: HashMap<PeerId, ConfigGenParams> =
311 local_config_gen_params(&peers, process_mgr.globals.FM_FEDERATION_BASE_PORT)?;
312
313 let mut admin_clients: BTreeMap<PeerId, DynGlobalApi> = BTreeMap::new();
314 let mut endpoints: BTreeMap<PeerId, _> = BTreeMap::new();
315 for peer_id in num_peers.peer_ids() {
316 let peer_env_vars = vars::Fedimintd::init(
317 &process_mgr.globals,
318 federation_name.clone(),
319 peer_id,
320 process_mgr
321 .globals
322 .fedimintd_overrides
323 .peer_expect(fed_index, peer_id),
324 )
325 .await?;
326 members.insert(
327 peer_id.to_usize(),
328 Fedimintd::new(
329 process_mgr,
330 bitcoind.clone(),
331 peer_id.to_usize(),
332 &peer_env_vars,
333 federation_name.clone(),
334 )
335 .await?,
336 );
337 let admin_client = DynGlobalApi::from_setup_endpoint(
338 SafeUrl::parse(&peer_env_vars.FM_API_URL)?,
339 &process_mgr.globals.FM_FORCE_API_SECRETS.get_active(),
340 )
341 .await?;
342 endpoints.insert(peer_id, peer_env_vars.FM_API_URL.clone());
343 admin_clients.insert(peer_id, admin_client);
344 peer_to_env_vars_map.insert(peer_id.to_usize(), peer_env_vars);
345 }
346
347 if !skip_setup && !pre_dkg {
348 let (original_fedimint_cli_path, original_fm_mint_client) =
351 crate::util::use_matching_fedimint_cli_for_dkg().await?;
352
353 let fedimint_cli_version = crate::util::FedimintCli::version_or_default().await;
354
355 if fedimint_cli_version >= *VERSION_0_7_0_ALPHA {
356 run_cli_dkg_v2(params, endpoints).await?;
357 } else {
358 run_cli_dkg(params, endpoints).await?;
359 }
360
361 crate::util::use_fedimint_cli(original_fedimint_cli_path, original_fm_mint_client);
363
364 let client_dir = utf8(&process_mgr.globals.FM_CLIENT_DIR);
366 let invite_code_filename_original = "invite-code";
367
368 for peer_env_vars in peer_to_env_vars_map.values() {
369 let peer_data_dir = utf8(&peer_env_vars.FM_DATA_DIR);
370
371 let invite_code = poll_simple("awaiting-invite-code", || async {
372 let path = format!("{peer_data_dir}/{invite_code_filename_original}");
373 tokio::fs::read_to_string(&path)
374 .await
375 .with_context(|| format!("Awaiting invite code file: {path}"))
376 })
377 .await
378 .context("Awaiting invite code file")?;
379
380 Connector::default()
381 .download_from_invite_code(&InviteCode::from_str(&invite_code)?)
382 .await?;
383 }
384
385 let peer_data_dir = utf8(&peer_to_env_vars_map[&0].FM_DATA_DIR);
387
388 tokio::fs::copy(
389 format!("{peer_data_dir}/{invite_code_filename_original}"),
390 format!("{client_dir}/{invite_code_filename_original}"),
391 )
392 .await
393 .context("copying invite-code file")?;
394
395 for (index, peer_env_vars) in &peer_to_env_vars_map {
398 let peer_data_dir = utf8(&peer_env_vars.FM_DATA_DIR);
399
400 let invite_code_filename_indexed =
401 format!("{invite_code_filename_original}-{index}");
402 tokio::fs::rename(
403 format!("{peer_data_dir}/{invite_code_filename_original}"),
404 format!("{client_dir}/{invite_code_filename_indexed}"),
405 )
406 .await
407 .context("moving invite-code file")?;
408 }
409
410 debug!("Moved invite-code files to client data directory");
411 }
412
413 let client = JitTryAnyhow::new_try({
414 move || async move {
415 let client = Client::open_or_create(federation_name.as_str())?;
416 let invite_code = Self::invite_code_static()?;
417 if !skip_setup && !pre_dkg {
418 cmd!(client, "join-federation", invite_code).run().await?;
419 }
420 Ok(client)
421 }
422 });
423
424 Ok(Self {
425 members,
426 vars: peer_to_env_vars_map,
427 bitcoind,
428 client,
429 })
430 }
431
432 pub fn client_config(&self) -> Result<ClientConfig> {
433 let cfg_path = self.vars[&0].FM_DATA_DIR.join("client.json");
434 load_from_file(&cfg_path)
435 }
436
437 pub fn module_client_config<M: ClientModule>(
438 &self,
439 ) -> Result<Option<<M::Common as ModuleCommon>::ClientConfig>> {
440 self.client_config()?
441 .modules
442 .iter()
443 .find_map(|(module_instance_id, module_cfg)| {
444 if module_cfg.kind == M::kind() {
445 let decoders = ModuleDecoderRegistry::new(vec![(
446 *module_instance_id,
447 M::kind(),
448 M::decoder(),
449 )]);
450 Some(
451 module_cfg
452 .config
453 .clone()
454 .redecode_raw(&decoders)
455 .expect("Decoding client cfg failed")
456 .expect_decoded_ref()
457 .as_any()
458 .downcast_ref::<<M::Common as ModuleCommon>::ClientConfig>()
459 .cloned()
460 .context("Cast to module config failed"),
461 )
462 } else {
463 None
464 }
465 })
466 .transpose()
467 }
468
469 pub fn deposit_fees(&self) -> Result<Amount> {
470 Ok(self
471 .module_client_config::<WalletClientModule>()?
472 .context("No wallet module found")?
473 .fee_consensus
474 .peg_in_abs)
475 }
476
477 pub fn invite_code(&self) -> Result<String> {
479 let data_dir: PathBuf = env::var(FM_CLIENT_DIR_ENV)?.parse()?;
480 let invite_code = fs::read_to_string(data_dir.join("invite-code"))?;
481 Ok(invite_code)
482 }
483
484 pub fn invite_code_static() -> Result<String> {
485 let data_dir: PathBuf = env::var(FM_CLIENT_DIR_ENV)?.parse()?;
486 let invite_code = fs::read_to_string(data_dir.join("invite-code"))?;
487 Ok(invite_code)
488 }
489 pub fn invite_code_for(peer_id: PeerId) -> Result<String> {
490 let data_dir: PathBuf = env::var(FM_CLIENT_DIR_ENV)?.parse()?;
491 let name = format!("invite-code-{peer_id}");
492 let invite_code = fs::read_to_string(data_dir.join(name))?;
493 Ok(invite_code)
494 }
495
496 pub async fn internal_client(&self) -> Result<&Client> {
500 self.client
501 .get_try()
502 .await
503 .context("Internal client joining Federation")
504 }
505
506 pub async fn new_joined_client(&self, name: impl ToString) -> Result<Client> {
508 let client = Client::create(name).await?;
509 client.join_federation(self.invite_code()?).await?;
510 Ok(client)
511 }
512
513 pub async fn start_server(&mut self, process_mgr: &ProcessManager, peer: usize) -> Result<()> {
514 if self.members.contains_key(&peer) {
515 bail!("fedimintd-{peer} already running");
516 }
517 self.members.insert(
518 peer,
519 Fedimintd::new(
520 process_mgr,
521 self.bitcoind.clone(),
522 peer,
523 &self.vars[&peer],
524 "default".to_string(),
525 )
526 .await?,
527 );
528 Ok(())
529 }
530
531 pub async fn terminate_server(&mut self, peer_id: usize) -> Result<()> {
532 let Some((_, fedimintd)) = self.members.remove_entry(&peer_id) else {
533 bail!("fedimintd-{peer_id} does not exist");
534 };
535 fedimintd.terminate().await?;
536 Ok(())
537 }
538
539 pub async fn start_all_servers(&mut self, process_mgr: &ProcessManager) -> Result<()> {
541 info!("starting all servers");
542 let fed_size = process_mgr.globals.FM_FED_SIZE;
543 for peer_id in 0..fed_size {
544 if self.members.contains_key(&peer_id) {
545 continue;
546 }
547 self.start_server(process_mgr, peer_id).await?;
548 }
549 self.await_all_peers().await?;
550 Ok(())
551 }
552
553 pub async fn terminate_all_servers(&mut self) -> Result<()> {
555 info!("terminating all servers");
556 let running_peer_ids: Vec<_> = self.members.keys().copied().collect();
557 for peer_id in running_peer_ids {
558 self.terminate_server(peer_id).await?;
559 }
560 Ok(())
561 }
562
563 pub async fn restart_all_staggered_with_bin(
568 &mut self,
569 process_mgr: &ProcessManager,
570 bin_path: &PathBuf,
571 ) -> Result<()> {
572 let fed_size = process_mgr.globals.FM_FED_SIZE;
573
574 self.start_all_servers(process_mgr).await?;
576
577 while self.num_members() > 0 {
579 self.terminate_server(self.num_members() - 1).await?;
580 if self.num_members() > 0 {
581 fedimint_core::task::sleep_in_test(
582 "waiting to shutdown remaining peers",
583 Duration::from_secs(10),
584 )
585 .await;
586 }
587 }
588
589 unsafe { std::env::set_var("FM_FEDIMINTD_BASE_EXECUTABLE", bin_path) };
591
592 for peer_id in 0..fed_size {
594 self.start_server(process_mgr, peer_id).await?;
595 if peer_id < fed_size - 1 {
596 fedimint_core::task::sleep_in_test(
597 "waiting to restart remaining peers",
598 Duration::from_secs(10),
599 )
600 .await;
601 }
602 }
603
604 self.await_all_peers().await?;
605
606 let fedimintd_version = crate::util::FedimintdCmd::version_or_default().await;
607 info!("upgraded fedimintd to version: {}", fedimintd_version);
608 Ok(())
609 }
610
611 pub async fn restart_all_with_bin(
612 &mut self,
613 process_mgr: &ProcessManager,
614 bin_path: &PathBuf,
615 ) -> Result<()> {
616 let current_fedimintd_path = std::env::var("FM_FEDIMINTD_BASE_EXECUTABLE")?;
618 unsafe { std::env::set_var("FM_FEDIMINTD_BASE_EXECUTABLE", bin_path) };
620 unsafe { std::env::set_var("FM_FEDIMINTD_BASE_EXECUTABLE", current_fedimintd_path) };
622
623 self.restart_all_staggered_with_bin(process_mgr, bin_path)
624 .await
625 }
626
627 pub async fn degrade_federation(&mut self, process_mgr: &ProcessManager) -> Result<()> {
628 let fed_size = process_mgr.globals.FM_FED_SIZE;
629 let offline_nodes = process_mgr.globals.FM_OFFLINE_NODES;
630 anyhow::ensure!(
631 fed_size > 3 * offline_nodes,
632 "too many offline nodes ({offline_nodes}) to reach consensus"
633 );
634
635 while self.num_members() > fed_size - offline_nodes {
636 self.terminate_server(self.num_members() - 1).await?;
637 }
638
639 if offline_nodes > 0 {
640 info!(fed_size, offline_nodes, "federation is degraded");
641 }
642 Ok(())
643 }
644
645 pub async fn pegin_client_no_wait(&self, amount: u64, client: &Client) -> Result<String> {
646 let deposit_fees_msat = self.deposit_fees()?.msats;
647 assert_eq!(
648 deposit_fees_msat % 1000,
649 0,
650 "Deposit fees expected to be whole sats in test suite"
651 );
652 let deposit_fees = deposit_fees_msat / 1000;
653 info!(amount, deposit_fees, "Pegging-in client funds");
654
655 let (address, operation_id) = client.get_deposit_addr().await?;
656
657 self.bitcoind
658 .send_to(address, amount + deposit_fees)
659 .await?;
660 self.bitcoind.mine_blocks(21).await?;
661
662 Ok(operation_id)
663 }
664
665 pub async fn pegin_client(&self, amount: u64, client: &Client) -> Result<()> {
666 let operation_id = self.pegin_client_no_wait(amount, client).await?;
667
668 client.await_deposit(&operation_id).await?;
669 Ok(())
670 }
671
672 pub async fn pegin_gateways(
675 &self,
676 amount: u64,
677 gateways: Vec<&super::gatewayd::Gatewayd>,
678 ) -> Result<()> {
679 let deposit_fees_msat = self.deposit_fees()?.msats;
680 assert_eq!(
681 deposit_fees_msat % 1000,
682 0,
683 "Deposit fees expected to be whole sats in test suite"
684 );
685 let deposit_fees = deposit_fees_msat / 1000;
686 info!(amount, deposit_fees, "Pegging-in gateway funds");
687 let fed_id = self.calculate_federation_id();
688 for gw in gateways.clone() {
689 let pegin_addr = gw.get_pegin_addr(&fed_id).await?;
690 self.bitcoind
691 .send_to(pegin_addr, amount + deposit_fees)
692 .await?;
693 }
694
695 self.bitcoind.mine_blocks(21).await?;
696 let bitcoind_block_height: u64 = self.bitcoind.get_block_count().await? - 1;
697 try_join_all(gateways.into_iter().map(|gw| {
698 poll("gateway pegin", || async {
699 let gw_info = gw.get_info().await.map_err(ControlFlow::Continue)?;
700 let block_height: u64 = gw_info["block_height"]
701 .as_u64()
702 .expect("Could not parse block height");
703 if bitcoind_block_height != block_height {
704 return Err(std::ops::ControlFlow::Continue(anyhow::anyhow!(
705 "gateway block height is not synced"
706 )));
707 }
708
709 let gateway_balance = gw
710 .ecash_balance(fed_id.clone())
711 .await
712 .map_err(ControlFlow::Continue)?;
713 poll_eq!(gateway_balance, amount * 1000)
714 })
715 }))
716 .await?;
717
718 Ok(())
719 }
720
721 pub async fn pegout_gateways(
724 &self,
725 amount: u64,
726 gateways: Vec<&super::gatewayd::Gatewayd>,
727 ) -> Result<()> {
728 info!(amount, "Pegging-out gateway funds");
729 let fed_id = self.calculate_federation_id();
730 let mut peg_outs: BTreeMap<LightningNodeType, (Amount, WithdrawResponse)> = BTreeMap::new();
731 for gw in gateways.clone() {
732 let prev_fed_ecash_balance = gw
733 .get_balances()
734 .await?
735 .ecash_balances
736 .into_iter()
737 .find(|fed| fed.federation_id.to_string() == fed_id)
738 .expect("Gateway has not joined federation")
739 .ecash_balance_msats;
740
741 let pegout_address = self.bitcoind.get_new_address().await?;
742 let value = cmd!(
743 gw,
744 "ecash",
745 "pegout",
746 "--federation-id",
747 fed_id,
748 "--amount",
749 amount,
750 "--address",
751 pegout_address
752 )
753 .out_json()
754 .await?;
755 let response: WithdrawResponse = serde_json::from_value(value)?;
756 peg_outs.insert(gw.ln.ln_type(), (prev_fed_ecash_balance, response));
757 }
758 self.bitcoind.mine_blocks(21).await?;
759
760 try_join_all(
761 peg_outs
762 .values()
763 .map(|(_, pegout)| self.bitcoind.poll_get_transaction(pegout.txid)),
764 )
765 .await?;
766
767 for gw in gateways.clone() {
768 let after_fed_ecash_balance = gw
769 .get_balances()
770 .await?
771 .ecash_balances
772 .into_iter()
773 .find(|fed| fed.federation_id.to_string() == fed_id)
774 .expect("Gateway has not joined federation")
775 .ecash_balance_msats;
776
777 let ln_type = gw.ln.ln_type();
778 let prev_balance = peg_outs
779 .get(&ln_type)
780 .expect("peg out does not exist")
781 .0
782 .msats;
783 let fees = peg_outs
784 .get(&ln_type)
785 .expect("peg out does not exist")
786 .1
787 .fees;
788 let total_fee = fees.amount().to_sat() * 1000;
789 assert_eq!(
790 prev_balance - amount - total_fee,
791 after_fed_ecash_balance.msats,
792 "new balance did not equal prev balance minus withdraw_amount minus fees"
793 );
794 }
795
796 Ok(())
797 }
798
799 pub fn calculate_federation_id(&self) -> String {
800 self.client_config()
801 .unwrap()
802 .global
803 .calculate_federation_id()
804 .to_string()
805 }
806
807 pub async fn await_block_sync(&self) -> Result<u64> {
808 let finality_delay = self.get_finality_delay()?;
809 let block_count = self.bitcoind.get_block_count().await?;
810 let expected = block_count.saturating_sub(finality_delay.into());
811 cmd!(
812 self.internal_client().await?,
813 "dev",
814 "wait-block-count",
815 expected
816 )
817 .run()
818 .await?;
819 Ok(expected)
820 }
821
822 fn get_finality_delay(&self) -> Result<u32, anyhow::Error> {
823 let client_config = &self.client_config()?;
824 let wallet_cfg = client_config
825 .modules
826 .get(&LEGACY_HARDCODED_INSTANCE_ID_WALLET)
827 .context("wallet module not found")?
828 .clone()
829 .redecode_raw(&ModuleDecoderRegistry::new([(
830 LEGACY_HARDCODED_INSTANCE_ID_WALLET,
831 fedimint_wallet_client::KIND,
832 fedimint_wallet_client::WalletModuleTypes::decoder(),
833 )]))?;
834 let wallet_cfg: &WalletClientConfig = wallet_cfg.cast()?;
835
836 let finality_delay = wallet_cfg.finality_delay;
837 Ok(finality_delay)
838 }
839
840 pub async fn await_gateways_registered(&self) -> Result<()> {
841 let start_time = Instant::now();
842 debug!(target: LOG_DEVIMINT, "Awaiting LN gateways registration");
843
844 poll("gateways registered", || async {
845 let num_gateways = cmd!(
846 self.internal_client()
847 .await
848 .map_err(ControlFlow::Continue)?,
849 "list-gateways"
850 )
851 .out_json()
852 .await
853 .map_err(ControlFlow::Continue)?
854 .as_array()
855 .context("invalid output")
856 .map_err(ControlFlow::Break)?
857 .len();
858 poll_eq!(num_gateways, 1)
859 })
860 .await?;
861 debug!(target: LOG_DEVIMINT,
862 elapsed_ms = %start_time.elapsed().as_millis(),
863 "Gateways registered");
864 Ok(())
865 }
866
867 pub async fn await_all_peers(&self) -> Result<()> {
868 poll("Waiting for all peers to be online", || async {
869 cmd!(
870 self.internal_client()
871 .await
872 .map_err(ControlFlow::Continue)?,
873 "dev",
874 "api",
875 "--module",
876 LEGACY_HARDCODED_INSTANCE_ID_WALLET,
877 "block_count"
878 )
879 .run()
880 .await
881 .map_err(ControlFlow::Continue)?;
882 Ok(())
883 })
884 .await
885 }
886
887 pub async fn finalize_mempool_tx(&self) -> Result<()> {
897 let finality_delay = self.get_finality_delay()?;
898 let blocks_to_mine = finality_delay + 1;
899 self.bitcoind.mine_blocks(blocks_to_mine.into()).await?;
900 self.await_block_sync().await?;
901 Ok(())
902 }
903
904 pub async fn mine_then_wait_blocks_sync(&self, blocks: u64) -> Result<()> {
905 self.bitcoind.mine_blocks(blocks).await?;
906 self.await_block_sync().await?;
907 Ok(())
908 }
909
910 pub fn num_members(&self) -> usize {
911 self.members.len()
912 }
913
914 pub fn member_ids(&self) -> impl Iterator<Item = PeerId> + '_ {
915 self.members
916 .keys()
917 .map(|&peer_id| PeerId::from(peer_id as u16))
918 }
919}
920
921#[derive(Clone)]
922pub struct Fedimintd {
923 _bitcoind: Bitcoind,
924 process: ProcessHandle,
925}
926
927impl Fedimintd {
928 pub async fn new(
929 process_mgr: &ProcessManager,
930 bitcoind: Bitcoind,
931 peer_id: usize,
932 env: &vars::Fedimintd,
933 fed_name: String,
934 ) -> Result<Self> {
935 debug!(target: LOG_DEVIMINT, "Starting fedimintd-{fed_name}-{peer_id}");
936 let process = process_mgr
937 .spawn_daemon(
938 &format!("fedimintd-{fed_name}-{peer_id}"),
939 cmd!(FedimintdCmd).envs(env.vars()),
940 )
941 .await?;
942
943 Ok(Self {
944 _bitcoind: bitcoind,
945 process,
946 })
947 }
948
949 pub async fn terminate(self) -> Result<()> {
950 self.process.terminate().await
951 }
952}
953
954pub async fn run_cli_dkg(
955 params: HashMap<PeerId, ConfigGenParams>,
956 endpoints: BTreeMap<PeerId, String>,
957) -> Result<()> {
958 let auth_for = |peer: &PeerId| -> &ApiAuth { ¶ms[peer].api_auth };
959
960 debug!(target: LOG_DEVIMINT, "Running DKG");
961 for endpoint in endpoints.values() {
962 poll("trying-to-connect-to-peers", || async {
963 crate::util::FedimintCli
964 .ws_status(endpoint)
965 .await
966 .context("dkg status")
967 .map_err(ControlFlow::Continue)
968 })
969 .await?;
970 }
971
972 debug!(target: LOG_DEVIMINT, "Connected to all peers");
973
974 for (peer_id, endpoint) in &endpoints {
975 let status = crate::util::FedimintCli.ws_status(endpoint).await?;
976 assert_eq!(
977 status.server,
978 ServerStatusLegacy::AwaitingPassword,
979 "peer_id isn't waiting for password: {peer_id}"
980 );
981 }
982
983 debug!(target: LOG_DEVIMINT, "Setting passwords");
984 for (peer_id, endpoint) in &endpoints {
985 crate::util::FedimintCli
986 .set_password(auth_for(peer_id), endpoint)
987 .await?;
988 }
989 let (leader_id, leader_endpoint) = endpoints.first_key_value().context("missing peer")?;
990 let followers = endpoints
991 .iter()
992 .filter(|(id, _)| *id != leader_id)
993 .collect::<BTreeMap<_, _>>();
994
995 debug!(target: LOG_DEVIMINT, "calling set_config_gen_connections for leader");
996 let leader_name = "leader".to_string();
997 crate::util::FedimintCli
998 .set_config_gen_connections(auth_for(leader_id), leader_endpoint, &leader_name, None)
999 .await?;
1000
1001 let server_gen_params = ServerModuleConfigGenParamsRegistry::default();
1002
1003 debug!(target: LOG_DEVIMINT, "calling set_config_gen_params for leader");
1004 cli_set_config_gen_params(
1005 leader_endpoint,
1006 auth_for(leader_id),
1007 server_gen_params.clone(),
1008 )
1009 .await?;
1010
1011 let followers_names = followers
1012 .keys()
1013 .map(|peer_id| {
1014 (*peer_id, {
1015 let random_string = rand::thread_rng()
1017 .sample_iter(&rand::distributions::Alphanumeric)
1018 .take(5)
1019 .map(char::from)
1020 .collect::<String>();
1021 format!("random-{random_string}{peer_id}")
1022 })
1023 })
1024 .collect::<BTreeMap<_, _>>();
1025 for (peer_id, endpoint) in &followers {
1026 let name = followers_names
1027 .get(peer_id)
1028 .context("missing follower name")?;
1029 debug!(target: LOG_DEVIMINT, "calling set_config_gen_connections for {peer_id} {name}");
1030
1031 crate::util::FedimintCli
1032 .set_config_gen_connections(auth_for(peer_id), endpoint, name, Some(leader_endpoint))
1033 .await?;
1034
1035 cli_set_config_gen_params(endpoint, auth_for(peer_id), server_gen_params.clone()).await?;
1036 }
1037
1038 debug!(target: LOG_DEVIMINT, "calling get_config_gen_peers for leader");
1039 let peers = crate::util::FedimintCli
1040 .get_config_gen_peers(leader_endpoint)
1041 .await?;
1042
1043 let found_names = peers
1044 .into_iter()
1045 .map(|peer| peer.name)
1046 .collect::<HashSet<_>>();
1047 let all_names = followers_names
1048 .values()
1049 .cloned()
1050 .chain(iter::once(leader_name))
1051 .collect::<HashSet<_>>();
1052 assert_eq!(found_names, all_names);
1053
1054 debug!(target: LOG_DEVIMINT, "Waiting for SharingConfigGenParams");
1055 cli_wait_server_status(leader_endpoint, ServerStatusLegacy::SharingConfigGenParams).await?;
1056
1057 debug!(target: LOG_DEVIMINT, "Getting consensus configs");
1058 let mut configs = vec![];
1059 for endpoint in endpoints.values() {
1060 let config = crate::util::FedimintCli
1061 .consensus_config_gen_params_legacy(endpoint)
1062 .await?;
1063 configs.push(config);
1064 }
1065 let mut consensus: Vec<_> = configs.iter().map(|p| p.consensus.clone()).collect();
1067 consensus.dedup();
1068 assert_eq!(consensus.len(), 1);
1069 let ids = configs
1071 .iter()
1072 .map(|p| p.our_current_id)
1073 .collect::<HashSet<_>>();
1074 assert_eq!(ids.len(), endpoints.len());
1075 let dkg_results = endpoints
1076 .iter()
1077 .map(|(peer_id, endpoint)| crate::util::FedimintCli.run_dkg(auth_for(peer_id), endpoint));
1078 debug!(target: LOG_DEVIMINT, "Running DKG");
1079 let (dkg_results, leader_wait_result) = tokio::join!(
1080 join_all(dkg_results),
1081 cli_wait_server_status(leader_endpoint, ServerStatusLegacy::VerifyingConfigs)
1082 );
1083 for result in dkg_results {
1084 result?;
1085 }
1086 leader_wait_result?;
1087
1088 debug!(target: LOG_DEVIMINT, "Verifying config hashes");
1090 let mut hashes = HashSet::new();
1091 for (peer_id, endpoint) in &endpoints {
1092 cli_wait_server_status(endpoint, ServerStatusLegacy::VerifyingConfigs).await?;
1093 let hash = crate::util::FedimintCli
1094 .get_verify_config_hash(auth_for(peer_id), endpoint)
1095 .await?;
1096 hashes.insert(hash);
1097 }
1098 assert_eq!(hashes.len(), 1);
1099 for (peer_id, endpoint) in &endpoints {
1100 let result = crate::util::FedimintCli
1101 .start_consensus(auth_for(peer_id), endpoint)
1102 .await;
1103 if let Err(e) = result {
1104 tracing::debug!(target: LOG_DEVIMINT, "Error calling start_consensus: {e:?}, trying to continue...");
1105 }
1106 cli_wait_server_status(endpoint, ServerStatusLegacy::ConsensusRunning).await?;
1107 }
1108 Ok(())
1109}
1110
1111pub async fn run_cli_dkg_v2(
1112 params: HashMap<PeerId, ConfigGenParams>,
1113 endpoints: BTreeMap<PeerId, String>,
1114) -> Result<()> {
1115 let auth_for = |peer: &PeerId| -> &ApiAuth { ¶ms[peer].api_auth };
1116
1117 for (peer, endpoint) in &endpoints {
1118 let status = poll("awaiting-setup-status-awaiting-local-params", || async {
1119 crate::util::FedimintCli
1120 .setup_status(auth_for(peer), endpoint)
1121 .await
1122 .map_err(ControlFlow::Continue)
1123 })
1124 .await
1125 .unwrap();
1126
1127 assert_eq!(status, SetupStatus::AwaitingLocalParams);
1128 }
1129
1130 debug!(target: LOG_DEVIMINT, "Setting local parameters...");
1131
1132 let mut connection_info = BTreeMap::new();
1133
1134 for (peer, endpoint) in &endpoints {
1135 let info = if peer.to_usize() == 0 {
1136 crate::util::FedimintCli
1137 .set_local_params_leader(peer, auth_for(peer), endpoint)
1138 .await?
1139 } else {
1140 crate::util::FedimintCli
1141 .set_local_params_follower(peer, auth_for(peer), endpoint)
1142 .await?
1143 };
1144
1145 connection_info.insert(peer, info);
1146 }
1147
1148 debug!(target: LOG_DEVIMINT, "Exchanging peer connection info...");
1149
1150 for (peer, info) in connection_info {
1151 for (p, endpoint) in &endpoints {
1152 if p != peer {
1153 crate::util::FedimintCli
1154 .add_peer(&info, auth_for(p), endpoint)
1155 .await?;
1156 }
1157 }
1158 }
1159
1160 debug!(target: LOG_DEVIMINT, "Starting DKG...");
1161
1162 for (peer, endpoint) in &endpoints {
1163 crate::util::FedimintCli
1164 .start_dkg(auth_for(peer), endpoint)
1165 .await?;
1166 }
1167
1168 Ok(())
1169}
1170
1171async fn cli_set_config_gen_params(
1172 endpoint: &str,
1173 auth: &ApiAuth,
1174 mut server_gen_params: ServerModuleConfigGenParamsRegistry,
1175) -> Result<()> {
1176 self::config::attach_default_module_init_params(
1177 &BitcoinRpcConfig::get_defaults_from_env_vars()?,
1178 &mut server_gen_params,
1179 Network::Regtest,
1180 10,
1181 );
1182
1183 let meta = iter::once(("federation_name".to_string(), "testfed".to_string())).collect();
1184
1185 crate::util::FedimintCli
1186 .set_config_gen_params(auth, endpoint, meta, server_gen_params)
1187 .await?;
1188
1189 Ok(())
1190}
1191
1192async fn cli_wait_server_status(endpoint: &str, expected_status: ServerStatusLegacy) -> Result<()> {
1193 poll(
1194 &format!("waiting-server-status: {expected_status:?}"),
1195 || async {
1196 let server_status = crate::util::FedimintCli
1197 .ws_status(endpoint)
1198 .await
1199 .context("server status")
1200 .map_err(ControlFlow::Continue)?
1201 .server;
1202 if server_status == expected_status {
1203 Ok(())
1204 } else {
1205 Err(ControlFlow::Continue(anyhow!(
1206 "expected status: {expected_status:?} current status: {server_status:?}"
1207 )))
1208 }
1209 },
1210 )
1211 .await?;
1212 Ok(())
1213}