/* This file is part of DarkFi (https://dark.fi)
*
* Copyright (C) 2020-2023 Dyne.org foundation
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see .
*/
//! Seed sync session creates a connection to the seed nodes specified in settings.
//! A new seed sync session is created every time we call [`P2p::start()`]. The
//! seed sync session loops through all the configured seeds and tries to connect
//! to them using a [`Connector`]. Seed sync either connects successfully, fails
//! with an error, or times out.
//!
//! If a seed node connects successfully, it runs a version exchange protocol,
//! stores the channel in the p2p list of channels, and discoonnects, removing
//! the channel from the channel list.
//!
//! The channel is registered using the [`Session::register_channel()`] trait
//! method. This invokes the Protocol Registry method `attach()`. Usually this
//! returns a list of protocols that we loop through and start. In this case,
//! `attach()` uses the bitflag selector to identify seed sessions and exclude
//! them.
//!
//! The version exchange occurs inside `register_channel()`. We create a handshake
//! task that runs the version exchange with the `perform_handshake_protocols()`
//! function. This runs the version exchange protocol, stores the channel in the
//! p2p list of channels, and subscribes to a stop signal.
use async_std::sync::{Arc, Weak};
use async_trait::async_trait;
use futures::future::join_all;
use log::{debug, info, warn};
use smol::Executor;
use url::Url;
use super::{
super::{
connector::Connector,
p2p::{DnetInfo, P2p, P2pPtr},
},
Session, SessionBitFlag, SESSION_SEED,
};
use crate::Result;
pub type SeedSyncSessionPtr = Arc;
/// Defines seed connections session
pub struct SeedSyncSession {
p2p: Weak,
}
impl SeedSyncSession {
/// Create a new seed sync session instance
pub fn new(p2p: Weak) -> SeedSyncSessionPtr {
Arc::new(Self { p2p })
}
/// Start the seed sync session. Creates a new task for every seed
/// connection and starts the seed on each task.
pub async fn start(self: Arc, executor: Arc>) -> Result<()> {
debug!(target: "net::session::seedsync_session", "SeedSyncSession::start() [START]");
let settings = self.p2p().settings();
if settings.seeds.is_empty() {
warn!(
target: "net::session::seedsync_session",
"[P2P] Skipping seed sync process since no seeds are configured.",
);
return Ok(())
}
// Gather tasks so we can execute concurrently
let mut tasks = Vec::with_capacity(settings.seeds.len());
for (i, seed) in settings.seeds.iter().enumerate() {
let ex_ = executor.clone();
let self_ = self.clone();
tasks.push(async move {
if let Err(e) = self_.clone().start_seed(i, seed.clone(), ex_.clone()).await {
warn!(
target: "net::session::seedsync_session",
"[P2P] Seed #{} connection failed: {}", i, e,
);
}
});
}
// Poll concurrently
join_all(tasks).await;
// Seed process complete
if self.p2p().hosts().is_empty().await {
warn!(target: "net::session::seedsync_session", "[P2P] Hosts pool empty after seeding");
}
debug!(target: "net::session::seedsync_session", "SeedSyncSession::start() [END]");
Ok(())
}
/// Connects to a seed socket address
async fn start_seed(
self: Arc,
seed_index: usize,
seed: Url,
ex: Arc>,
) -> Result<()> {
debug!(
target: "net::session::seedsync_session", "SeedSyncSession::start_seed(i={}) [START]",
seed_index
);
let settings = self.p2p.upgrade().unwrap().settings();
let parent = Arc::downgrade(&self);
let connector = Connector::new(settings.clone(), Arc::new(parent));
match connector.connect(&seed).await {
Ok((url, ch)) => {
info!(
target: "net::session::seedsync_session",
"[P2P] Connected seed #{} [{}]", seed_index, url,
);
if let Err(e) = self.clone().register_channel(ch.clone(), ex.clone()).await {
warn!(
target: "net::session::seedsync_session",
"[P2P] Failure during sync seed session #{} [{}]: {}",
seed_index, url, e,
);
}
info!(
target: "net::session::seedsync_session",
"[P2P] Disconnecting from seed #{} [{}]",
seed_index, url,
);
ch.stop().await;
}
Err(e) => {
warn!(
target: "net::session:seedsync_session",
"[P2P] Failure contacting seed #{} [{}]: {}",
seed_index, seed, e
);
return Err(e)
}
}
debug!(
target: "net::session::seedsync_session",
"SeedSyncSession::start_seed(i={}) [END]",
seed_index
);
Ok(())
}
}
#[async_trait]
impl Session for SeedSyncSession {
fn p2p(&self) -> P2pPtr {
self.p2p.upgrade().unwrap()
}
fn type_id(&self) -> SessionBitFlag {
SESSION_SEED
}
async fn dnet_info(&self) -> DnetInfo {
todo!()
}
}