/* This file is part of DarkFi (https://dark.fi)
*
* Copyright (C) 2020-2024 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 .
*/
use std::{
sync::{Arc, Weak},
time::UNIX_EPOCH,
};
use async_trait::async_trait;
use log::debug;
use smol::Executor;
use super::{channel::ChannelPtr, hosts::HostColor, p2p::P2pPtr, protocol::ProtocolVersion};
use crate::Result;
pub mod inbound_session;
pub use inbound_session::{InboundSession, InboundSessionPtr};
pub mod manual_session;
pub use manual_session::{ManualSession, ManualSessionPtr};
pub mod outbound_session;
pub use outbound_session::{OutboundSession, OutboundSessionPtr};
pub mod seedsync_session;
pub use seedsync_session::{SeedSyncSession, SeedSyncSessionPtr};
pub mod refine_session;
pub use refine_session::{RefineSession, RefineSessionPtr};
/// Bitwise selectors for the `protocol_registry`
pub type SessionBitFlag = u32;
pub const SESSION_INBOUND: SessionBitFlag = 0b00001;
pub const SESSION_OUTBOUND: SessionBitFlag = 0b00010;
pub const SESSION_MANUAL: SessionBitFlag = 0b00100;
pub const SESSION_SEED: SessionBitFlag = 0b01000;
pub const SESSION_REFINE: SessionBitFlag = 0b10000;
pub const SESSION_DEFAULT: SessionBitFlag = 0b00111;
pub const SESSION_ALL: SessionBitFlag = 0b11111;
pub type SessionWeakPtr = Weak;
/// Removes channel from the list of connected channels when a stop signal
/// is received.
pub async fn remove_sub_on_stop(p2p: P2pPtr, channel: ChannelPtr, type_id: SessionBitFlag) {
debug!(target: "net::session::remove_sub_on_stop()", "[START]");
let addr = channel.address();
// Subscribe to stop events
let stop_sub = channel.clone().subscribe_stop().await;
if let Ok(stop_sub) = stop_sub {
// Wait for a stop event
stop_sub.receive().await;
}
debug!(
target: "net::session::remove_sub_on_stop()",
"Received stop event. Removing channel {}", addr,
);
if type_id == SESSION_MANUAL || type_id == SESSION_OUTBOUND {
debug!(
target: "net::session::remove_sub_on_stop()",
"Downgrading {}", addr,
);
let last_seen = p2p.hosts().fetch_last_seen(addr).await.unwrap();
p2p.hosts().move_host(addr, last_seen, HostColor::Grey).await.unwrap();
}
// Remove channel from the HostRegistry. Free up this addr for any future operation.
p2p.hosts().unregister(channel.address()).await;
debug!(target: "net::session::remove_sub_on_stop()", "[END]");
}
/// Session trait. Defines methods that are used across sessions.
/// Implements registering the channel and initializing the channel by
/// performing a network handshake.
#[async_trait]
pub trait Session: Sync {
/// Registers a new channel with the session.
/// Performs a network handshake and starts the channel.
/// If we need to pass `Self` as an `Arc` we can do so like this:
/// ```
/// pub trait MyTrait: Send + Sync {
/// async fn foo(&self, self_: Arc) {}
/// }
/// ```
async fn register_channel(
&self,
channel: ChannelPtr,
executor: Arc>,
) -> Result<()> {
debug!(target: "net::session::register_channel()", "[START]");
// Protocols should all be initialized but not started.
// We do this so that the protocols can begin receiving and buffering
// messages while the handshake protocol is ongoing. They are currently
// in sleep mode.
let p2p = self.p2p();
let protocols =
p2p.protocol_registry().attach(self.type_id(), channel.clone(), p2p.clone()).await;
// Perform the handshake protocol
let protocol_version = ProtocolVersion::new(channel.clone(), p2p.settings().clone()).await;
debug!(target: "net::session::register_channel()",
"Performing handshake protocols {}", channel.clone().address());
let handshake_task =
self.perform_handshake_protocols(protocol_version, channel.clone(), executor.clone());
// Switch on the channel
channel.clone().start(executor.clone());
// Wait for handshake to finish.
// If the handshake returns an error, remove this host from all hostlists.
match handshake_task.await {
Ok(()) => {
debug!(target: "net::session::register_channel()",
"Handshake successful {}", channel.clone().address());
}
Err(e) => {
debug!(target: "net::session::register_channel()",
"Handshake error {} {}", e, channel.clone().address());
}
}
// Now the channel is ready
debug!(target: "net::session::register_channel()", "Session handshake complete");
debug!(target: "net::session::register_channel()", "Activating remaining protocols");
// Now start all the protocols. They are responsible for managing their own
// lifetimes and correctly selfdestructing when the channel ends.
for protocol in protocols {
protocol.start(executor.clone()).await?;
}
debug!(target: "net::session::register_channel()", "[END]");
Ok(())
}
/// Performs network handshake to initialize channel. Adds the channel to
/// the list of connected channels, and prepares to remove the channel when
/// a stop signal is received.
async fn perform_handshake_protocols(
&self,
protocol_version: Arc,
channel: ChannelPtr,
executor: Arc>,
) -> Result<()> {
// Perform handshake
protocol_version.run(executor.clone()).await?;
// Upgrade to goldlist if outbound or manual session.
if self.type_id() == SESSION_MANUAL || self.type_id() == SESSION_OUTBOUND {
debug!(
target: "net::session::perform_handshake_protocols()",
"Upgrading {}", channel.address(),
);
let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
self.p2p()
.hosts()
.move_host(channel.address(), last_seen, HostColor::Gold)
.await
.unwrap();
}
// Attempt to add channel to registry
self.p2p().hosts().register_channel(channel.clone()).await;
// Subscribe to stop, so we can remove from registry
executor.spawn(remove_sub_on_stop(self.p2p(), channel, self.type_id())).detach();
// Channel is ready for use
Ok(())
}
/// Returns a pointer to the p2p network interface
fn p2p(&self) -> P2pPtr;
/// Return the session bit flag for the session type
fn type_id(&self) -> SessionBitFlag;
}