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_almost_equal, 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 await_server_terminated(&mut self, peer_id: usize) -> Result<()> {
540 let Some(fedimintd) = self.members.get_mut(&peer_id) else {
541 bail!("fedimintd-{peer_id} does not exist");
542 };
543 fedimintd.await_terminated().await?;
544 self.members.remove(&peer_id);
545 Ok(())
546 }
547
548 pub async fn start_all_servers(&mut self, process_mgr: &ProcessManager) -> Result<()> {
550 info!("starting all servers");
551 let fed_size = process_mgr.globals.FM_FED_SIZE;
552 for peer_id in 0..fed_size {
553 if self.members.contains_key(&peer_id) {
554 continue;
555 }
556 self.start_server(process_mgr, peer_id).await?;
557 }
558 self.await_all_peers().await?;
559 Ok(())
560 }
561
562 pub async fn terminate_all_servers(&mut self) -> Result<()> {
564 info!("terminating all servers");
565 let running_peer_ids: Vec<_> = self.members.keys().copied().collect();
566 for peer_id in running_peer_ids {
567 self.terminate_server(peer_id).await?;
568 }
569 Ok(())
570 }
571
572 pub async fn restart_all_staggered_with_bin(
577 &mut self,
578 process_mgr: &ProcessManager,
579 bin_path: &PathBuf,
580 ) -> Result<()> {
581 let fed_size = process_mgr.globals.FM_FED_SIZE;
582
583 self.start_all_servers(process_mgr).await?;
585
586 while self.num_members() > 0 {
588 self.terminate_server(self.num_members() - 1).await?;
589 if self.num_members() > 0 {
590 fedimint_core::task::sleep_in_test(
591 "waiting to shutdown remaining peers",
592 Duration::from_secs(10),
593 )
594 .await;
595 }
596 }
597
598 unsafe { std::env::set_var("FM_FEDIMINTD_BASE_EXECUTABLE", bin_path) };
600
601 for peer_id in 0..fed_size {
603 self.start_server(process_mgr, peer_id).await?;
604 if peer_id < fed_size - 1 {
605 fedimint_core::task::sleep_in_test(
606 "waiting to restart remaining peers",
607 Duration::from_secs(10),
608 )
609 .await;
610 }
611 }
612
613 self.await_all_peers().await?;
614
615 let fedimintd_version = crate::util::FedimintdCmd::version_or_default().await;
616 info!("upgraded fedimintd to version: {}", fedimintd_version);
617 Ok(())
618 }
619
620 pub async fn restart_all_with_bin(
621 &mut self,
622 process_mgr: &ProcessManager,
623 bin_path: &PathBuf,
624 ) -> Result<()> {
625 let current_fedimintd_path = std::env::var("FM_FEDIMINTD_BASE_EXECUTABLE")?;
627 unsafe { std::env::set_var("FM_FEDIMINTD_BASE_EXECUTABLE", bin_path) };
629 unsafe { std::env::set_var("FM_FEDIMINTD_BASE_EXECUTABLE", current_fedimintd_path) };
631
632 self.restart_all_staggered_with_bin(process_mgr, bin_path)
633 .await
634 }
635
636 pub async fn degrade_federation(&mut self, process_mgr: &ProcessManager) -> Result<()> {
637 let fed_size = process_mgr.globals.FM_FED_SIZE;
638 let offline_nodes = process_mgr.globals.FM_OFFLINE_NODES;
639 anyhow::ensure!(
640 fed_size > 3 * offline_nodes,
641 "too many offline nodes ({offline_nodes}) to reach consensus"
642 );
643
644 while self.num_members() > fed_size - offline_nodes {
645 self.terminate_server(self.num_members() - 1).await?;
646 }
647
648 if offline_nodes > 0 {
649 info!(fed_size, offline_nodes, "federation is degraded");
650 }
651 Ok(())
652 }
653
654 pub async fn pegin_client_no_wait(&self, amount: u64, client: &Client) -> Result<String> {
655 let deposit_fees_msat = self.deposit_fees()?.msats;
656 assert_eq!(
657 deposit_fees_msat % 1000,
658 0,
659 "Deposit fees expected to be whole sats in test suite"
660 );
661 let deposit_fees = deposit_fees_msat / 1000;
662 info!(amount, deposit_fees, "Pegging-in client funds");
663
664 let (address, operation_id) = client.get_deposit_addr().await?;
665
666 self.bitcoind
667 .send_to(address, amount + deposit_fees)
668 .await?;
669 self.bitcoind.mine_blocks(21).await?;
670
671 Ok(operation_id)
672 }
673
674 pub async fn pegin_client(&self, amount: u64, client: &Client) -> Result<()> {
675 let operation_id = self.pegin_client_no_wait(amount, client).await?;
676
677 client.await_deposit(&operation_id).await?;
678 Ok(())
679 }
680
681 pub async fn pegin_gateways(
684 &self,
685 amount: u64,
686 gateways: Vec<&super::gatewayd::Gatewayd>,
687 ) -> Result<()> {
688 let deposit_fees_msat = self.deposit_fees()?.msats;
689 assert_eq!(
690 deposit_fees_msat % 1000,
691 0,
692 "Deposit fees expected to be whole sats in test suite"
693 );
694 let deposit_fees = deposit_fees_msat / 1000;
695 info!(amount, deposit_fees, "Pegging-in gateway funds");
696 let fed_id = self.calculate_federation_id();
697 for gw in gateways.clone() {
698 let pegin_addr = gw.get_pegin_addr(&fed_id).await?;
699 self.bitcoind
700 .send_to(pegin_addr, amount + deposit_fees)
701 .await?;
702 }
703
704 self.bitcoind.mine_blocks(21).await?;
705 let bitcoind_block_height: u64 = self.bitcoind.get_block_count().await? - 1;
706 try_join_all(gateways.into_iter().map(|gw| {
707 poll("gateway pegin", || async {
708 let gw_info = gw.get_info().await.map_err(ControlFlow::Continue)?;
709 let block_height: u64 = gw_info["block_height"]
710 .as_u64()
711 .expect("Could not parse block height");
712 if bitcoind_block_height != block_height {
713 return Err(std::ops::ControlFlow::Continue(anyhow::anyhow!(
714 "gateway block height is not synced"
715 )));
716 }
717
718 let gateway_balance = gw
719 .ecash_balance(fed_id.clone())
720 .await
721 .map_err(ControlFlow::Continue)?;
722 poll_almost_equal!(gateway_balance, amount * 1000)
723 })
724 }))
725 .await?;
726
727 Ok(())
728 }
729
730 pub async fn pegout_gateways(
733 &self,
734 amount: u64,
735 gateways: Vec<&super::gatewayd::Gatewayd>,
736 ) -> Result<()> {
737 info!(amount, "Pegging-out gateway funds");
738 let fed_id = self.calculate_federation_id();
739 let mut peg_outs: BTreeMap<LightningNodeType, (Amount, WithdrawResponse)> = BTreeMap::new();
740 for gw in gateways.clone() {
741 let prev_fed_ecash_balance = gw
742 .get_balances()
743 .await?
744 .ecash_balances
745 .into_iter()
746 .find(|fed| fed.federation_id.to_string() == fed_id)
747 .expect("Gateway has not joined federation")
748 .ecash_balance_msats;
749
750 let pegout_address = self.bitcoind.get_new_address().await?;
751 let value = cmd!(
752 gw,
753 "ecash",
754 "pegout",
755 "--federation-id",
756 fed_id,
757 "--amount",
758 amount,
759 "--address",
760 pegout_address
761 )
762 .out_json()
763 .await?;
764 let response: WithdrawResponse = serde_json::from_value(value)?;
765 peg_outs.insert(gw.ln.ln_type(), (prev_fed_ecash_balance, response));
766 }
767 self.bitcoind.mine_blocks(21).await?;
768
769 try_join_all(
770 peg_outs
771 .values()
772 .map(|(_, pegout)| self.bitcoind.poll_get_transaction(pegout.txid)),
773 )
774 .await?;
775
776 for gw in gateways.clone() {
777 let after_fed_ecash_balance = gw
778 .get_balances()
779 .await?
780 .ecash_balances
781 .into_iter()
782 .find(|fed| fed.federation_id.to_string() == fed_id)
783 .expect("Gateway has not joined federation")
784 .ecash_balance_msats;
785
786 let ln_type = gw.ln.ln_type();
787 let prev_balance = peg_outs
788 .get(&ln_type)
789 .expect("peg out does not exist")
790 .0
791 .msats;
792 let fees = peg_outs
793 .get(&ln_type)
794 .expect("peg out does not exist")
795 .1
796 .fees;
797 let total_fee = fees.amount().to_sat() * 1000;
798 crate::util::almost_equal(
799 after_fed_ecash_balance.msats,
800 prev_balance - amount - total_fee,
801 2000,
802 )
803 .map_err(|e| {
804 anyhow::anyhow!(
805 "new balance did not equal prev balance minus withdraw_amount minus fees: {}",
806 e
807 )
808 })?;
809 }
810
811 Ok(())
812 }
813
814 pub fn calculate_federation_id(&self) -> String {
815 self.client_config()
816 .unwrap()
817 .global
818 .calculate_federation_id()
819 .to_string()
820 }
821
822 pub async fn await_block_sync(&self) -> Result<u64> {
823 let finality_delay = self.get_finality_delay()?;
824 let block_count = self.bitcoind.get_block_count().await?;
825 let expected = block_count.saturating_sub(finality_delay.into());
826 cmd!(
827 self.internal_client().await?,
828 "dev",
829 "wait-block-count",
830 expected
831 )
832 .run()
833 .await?;
834 Ok(expected)
835 }
836
837 fn get_finality_delay(&self) -> Result<u32, anyhow::Error> {
838 let client_config = &self.client_config()?;
839 let wallet_cfg = client_config
840 .modules
841 .get(&LEGACY_HARDCODED_INSTANCE_ID_WALLET)
842 .context("wallet module not found")?
843 .clone()
844 .redecode_raw(&ModuleDecoderRegistry::new([(
845 LEGACY_HARDCODED_INSTANCE_ID_WALLET,
846 fedimint_wallet_client::KIND,
847 fedimint_wallet_client::WalletModuleTypes::decoder(),
848 )]))?;
849 let wallet_cfg: &WalletClientConfig = wallet_cfg.cast()?;
850
851 let finality_delay = wallet_cfg.finality_delay;
852 Ok(finality_delay)
853 }
854
855 pub async fn await_gateways_registered(&self) -> Result<()> {
856 let start_time = Instant::now();
857 debug!(target: LOG_DEVIMINT, "Awaiting LN gateways registration");
858
859 poll("gateways registered", || async {
860 let num_gateways = cmd!(
861 self.internal_client()
862 .await
863 .map_err(ControlFlow::Continue)?,
864 "list-gateways"
865 )
866 .out_json()
867 .await
868 .map_err(ControlFlow::Continue)?
869 .as_array()
870 .context("invalid output")
871 .map_err(ControlFlow::Break)?
872 .len();
873 poll_eq!(num_gateways, 1)
874 })
875 .await?;
876 debug!(target: LOG_DEVIMINT,
877 elapsed_ms = %start_time.elapsed().as_millis(),
878 "Gateways registered");
879 Ok(())
880 }
881
882 pub async fn await_all_peers(&self) -> Result<()> {
883 poll("Waiting for all peers to be online", || async {
884 cmd!(
885 self.internal_client()
886 .await
887 .map_err(ControlFlow::Continue)?,
888 "dev",
889 "api",
890 "--module",
891 LEGACY_HARDCODED_INSTANCE_ID_WALLET,
892 "block_count"
893 )
894 .run()
895 .await
896 .map_err(ControlFlow::Continue)?;
897 Ok(())
898 })
899 .await
900 }
901
902 pub async fn await_peer(&self, peer_id: usize) -> Result<()> {
903 poll("Waiting for all peers to be online", || async {
904 cmd!(
905 self.internal_client()
906 .await
907 .map_err(ControlFlow::Continue)?,
908 "dev",
909 "api",
910 "--peer-id",
911 peer_id,
912 "--module",
913 LEGACY_HARDCODED_INSTANCE_ID_WALLET,
914 "block_count"
915 )
916 .run()
917 .await
918 .map_err(ControlFlow::Continue)?;
919 Ok(())
920 })
921 .await
922 }
923
924 pub async fn finalize_mempool_tx(&self) -> Result<()> {
934 let finality_delay = self.get_finality_delay()?;
935 let blocks_to_mine = finality_delay + 1;
936 self.bitcoind.mine_blocks(blocks_to_mine.into()).await?;
937 self.await_block_sync().await?;
938 Ok(())
939 }
940
941 pub async fn mine_then_wait_blocks_sync(&self, blocks: u64) -> Result<()> {
942 self.bitcoind.mine_blocks(blocks).await?;
943 self.await_block_sync().await?;
944 Ok(())
945 }
946
947 pub fn num_members(&self) -> usize {
948 self.members.len()
949 }
950
951 pub fn member_ids(&self) -> impl Iterator<Item = PeerId> + '_ {
952 self.members
953 .keys()
954 .map(|&peer_id| PeerId::from(peer_id as u16))
955 }
956}
957
958#[derive(Clone)]
959pub struct Fedimintd {
960 _bitcoind: Bitcoind,
961 process: ProcessHandle,
962}
963
964impl Fedimintd {
965 pub async fn new(
966 process_mgr: &ProcessManager,
967 bitcoind: Bitcoind,
968 peer_id: usize,
969 env: &vars::Fedimintd,
970 fed_name: String,
971 ) -> Result<Self> {
972 debug!(target: LOG_DEVIMINT, "Starting fedimintd-{fed_name}-{peer_id}");
973 let process = process_mgr
974 .spawn_daemon(
975 &format!("fedimintd-{fed_name}-{peer_id}"),
976 cmd!(FedimintdCmd).envs(env.vars()),
977 )
978 .await?;
979
980 Ok(Self {
981 _bitcoind: bitcoind,
982 process,
983 })
984 }
985
986 pub async fn terminate(self) -> Result<()> {
987 self.process.terminate().await
988 }
989
990 pub async fn await_terminated(&self) -> Result<()> {
991 self.process.await_terminated().await
992 }
993}
994
995pub async fn run_cli_dkg(
996 params: HashMap<PeerId, ConfigGenParams>,
997 endpoints: BTreeMap<PeerId, String>,
998) -> Result<()> {
999 let auth_for = |peer: &PeerId| -> &ApiAuth { ¶ms[peer].api_auth };
1000
1001 debug!(target: LOG_DEVIMINT, "Running DKG");
1002 for endpoint in endpoints.values() {
1003 poll("trying-to-connect-to-peers", || async {
1004 crate::util::FedimintCli
1005 .ws_status(endpoint)
1006 .await
1007 .context("dkg status")
1008 .map_err(ControlFlow::Continue)
1009 })
1010 .await?;
1011 }
1012
1013 debug!(target: LOG_DEVIMINT, "Connected to all peers");
1014
1015 for (peer_id, endpoint) in &endpoints {
1016 let status = crate::util::FedimintCli.ws_status(endpoint).await?;
1017 assert_eq!(
1018 status.server,
1019 ServerStatusLegacy::AwaitingPassword,
1020 "peer_id isn't waiting for password: {peer_id}"
1021 );
1022 }
1023
1024 debug!(target: LOG_DEVIMINT, "Setting passwords");
1025 for (peer_id, endpoint) in &endpoints {
1026 crate::util::FedimintCli
1027 .set_password(auth_for(peer_id), endpoint)
1028 .await?;
1029 }
1030 let (leader_id, leader_endpoint) = endpoints.first_key_value().context("missing peer")?;
1031 let followers = endpoints
1032 .iter()
1033 .filter(|(id, _)| *id != leader_id)
1034 .collect::<BTreeMap<_, _>>();
1035
1036 debug!(target: LOG_DEVIMINT, "calling set_config_gen_connections for leader");
1037 let leader_name = "leader".to_string();
1038 crate::util::FedimintCli
1039 .set_config_gen_connections(auth_for(leader_id), leader_endpoint, &leader_name, None)
1040 .await?;
1041
1042 let server_gen_params = ServerModuleConfigGenParamsRegistry::default();
1043
1044 debug!(target: LOG_DEVIMINT, "calling set_config_gen_params for leader");
1045 cli_set_config_gen_params(
1046 leader_endpoint,
1047 auth_for(leader_id),
1048 server_gen_params.clone(),
1049 )
1050 .await?;
1051
1052 let followers_names = followers
1053 .keys()
1054 .map(|peer_id| {
1055 (*peer_id, {
1056 let random_string = rand::thread_rng()
1058 .sample_iter(&rand::distributions::Alphanumeric)
1059 .take(5)
1060 .map(char::from)
1061 .collect::<String>();
1062 format!("random-{random_string}{peer_id}")
1063 })
1064 })
1065 .collect::<BTreeMap<_, _>>();
1066 for (peer_id, endpoint) in &followers {
1067 let name = followers_names
1068 .get(peer_id)
1069 .context("missing follower name")?;
1070 debug!(target: LOG_DEVIMINT, "calling set_config_gen_connections for {peer_id} {name}");
1071
1072 crate::util::FedimintCli
1073 .set_config_gen_connections(auth_for(peer_id), endpoint, name, Some(leader_endpoint))
1074 .await?;
1075
1076 cli_set_config_gen_params(endpoint, auth_for(peer_id), server_gen_params.clone()).await?;
1077 }
1078
1079 debug!(target: LOG_DEVIMINT, "calling get_config_gen_peers for leader");
1080 let peers = crate::util::FedimintCli
1081 .get_config_gen_peers(leader_endpoint)
1082 .await?;
1083
1084 let found_names = peers
1085 .into_iter()
1086 .map(|peer| peer.name)
1087 .collect::<HashSet<_>>();
1088 let all_names = followers_names
1089 .values()
1090 .cloned()
1091 .chain(iter::once(leader_name))
1092 .collect::<HashSet<_>>();
1093 assert_eq!(found_names, all_names);
1094
1095 debug!(target: LOG_DEVIMINT, "Waiting for SharingConfigGenParams");
1096 cli_wait_server_status(leader_endpoint, ServerStatusLegacy::SharingConfigGenParams).await?;
1097
1098 debug!(target: LOG_DEVIMINT, "Getting consensus configs");
1099 let mut configs = vec![];
1100 for endpoint in endpoints.values() {
1101 let config = crate::util::FedimintCli
1102 .consensus_config_gen_params_legacy(endpoint)
1103 .await?;
1104 configs.push(config);
1105 }
1106 let mut consensus: Vec<_> = configs.iter().map(|p| p.consensus.clone()).collect();
1108 consensus.dedup();
1109 assert_eq!(consensus.len(), 1);
1110 let ids = configs
1112 .iter()
1113 .map(|p| p.our_current_id)
1114 .collect::<HashSet<_>>();
1115 assert_eq!(ids.len(), endpoints.len());
1116 let dkg_results = endpoints
1117 .iter()
1118 .map(|(peer_id, endpoint)| crate::util::FedimintCli.run_dkg(auth_for(peer_id), endpoint));
1119 debug!(target: LOG_DEVIMINT, "Running DKG");
1120 let (dkg_results, leader_wait_result) = tokio::join!(
1121 join_all(dkg_results),
1122 cli_wait_server_status(leader_endpoint, ServerStatusLegacy::VerifyingConfigs)
1123 );
1124 for result in dkg_results {
1125 result?;
1126 }
1127 leader_wait_result?;
1128
1129 debug!(target: LOG_DEVIMINT, "Verifying config hashes");
1131 let mut hashes = HashSet::new();
1132 for (peer_id, endpoint) in &endpoints {
1133 cli_wait_server_status(endpoint, ServerStatusLegacy::VerifyingConfigs).await?;
1134 let hash = crate::util::FedimintCli
1135 .get_verify_config_hash(auth_for(peer_id), endpoint)
1136 .await?;
1137 hashes.insert(hash);
1138 }
1139 assert_eq!(hashes.len(), 1);
1140 for (peer_id, endpoint) in &endpoints {
1141 let result = crate::util::FedimintCli
1142 .start_consensus(auth_for(peer_id), endpoint)
1143 .await;
1144 if let Err(e) = result {
1145 tracing::debug!(target: LOG_DEVIMINT, "Error calling start_consensus: {e:?}, trying to continue...");
1146 }
1147 cli_wait_server_status(endpoint, ServerStatusLegacy::ConsensusRunning).await?;
1148 }
1149 Ok(())
1150}
1151
1152pub async fn run_cli_dkg_v2(
1153 params: HashMap<PeerId, ConfigGenParams>,
1154 endpoints: BTreeMap<PeerId, String>,
1155) -> Result<()> {
1156 let auth_for = |peer: &PeerId| -> &ApiAuth { ¶ms[peer].api_auth };
1157
1158 for (peer, endpoint) in &endpoints {
1159 let status = poll("awaiting-setup-status-awaiting-local-params", || async {
1160 crate::util::FedimintCli
1161 .setup_status(auth_for(peer), endpoint)
1162 .await
1163 .map_err(ControlFlow::Continue)
1164 })
1165 .await
1166 .unwrap();
1167
1168 assert_eq!(status, SetupStatus::AwaitingLocalParams);
1169 }
1170
1171 debug!(target: LOG_DEVIMINT, "Setting local parameters...");
1172
1173 let mut connection_info = BTreeMap::new();
1174
1175 for (peer, endpoint) in &endpoints {
1176 let info = if peer.to_usize() == 0 {
1177 crate::util::FedimintCli
1178 .set_local_params_leader(peer, auth_for(peer), endpoint)
1179 .await?
1180 } else {
1181 crate::util::FedimintCli
1182 .set_local_params_follower(peer, auth_for(peer), endpoint)
1183 .await?
1184 };
1185
1186 connection_info.insert(peer, info);
1187 }
1188
1189 debug!(target: LOG_DEVIMINT, "Exchanging peer connection info...");
1190
1191 for (peer, info) in connection_info {
1192 for (p, endpoint) in &endpoints {
1193 if p != peer {
1194 crate::util::FedimintCli
1195 .add_peer(&info, auth_for(p), endpoint)
1196 .await?;
1197 }
1198 }
1199 }
1200
1201 debug!(target: LOG_DEVIMINT, "Starting DKG...");
1202
1203 for (peer, endpoint) in &endpoints {
1204 crate::util::FedimintCli
1205 .start_dkg(auth_for(peer), endpoint)
1206 .await?;
1207 }
1208
1209 Ok(())
1210}
1211
1212async fn cli_set_config_gen_params(
1213 endpoint: &str,
1214 auth: &ApiAuth,
1215 mut server_gen_params: ServerModuleConfigGenParamsRegistry,
1216) -> Result<()> {
1217 self::config::attach_default_module_init_params(
1218 &BitcoinRpcConfig::get_defaults_from_env_vars()?,
1219 &mut server_gen_params,
1220 Network::Regtest,
1221 10,
1222 );
1223
1224 let meta = iter::once(("federation_name".to_string(), "testfed".to_string())).collect();
1225
1226 crate::util::FedimintCli
1227 .set_config_gen_params(auth, endpoint, meta, server_gen_params)
1228 .await?;
1229
1230 Ok(())
1231}
1232
1233async fn cli_wait_server_status(endpoint: &str, expected_status: ServerStatusLegacy) -> Result<()> {
1234 poll(
1235 &format!("waiting-server-status: {expected_status:?}"),
1236 || async {
1237 let server_status = crate::util::FedimintCli
1238 .ws_status(endpoint)
1239 .await
1240 .context("server status")
1241 .map_err(ControlFlow::Continue)?
1242 .server;
1243 if server_status == expected_status {
1244 Ok(())
1245 } else {
1246 Err(ControlFlow::Continue(anyhow!(
1247 "expected status: {expected_status:?} current status: {server_status:?}"
1248 )))
1249 }
1250 },
1251 )
1252 .await?;
1253 Ok(())
1254}