tensamin/omikron
Watch
6
0
Fork
You've already forked omikron
 0
Code Issues Pull requests 3 Projects Releases Packages Wiki Activity Actions

[Fix] Removed so much now useless logging ._.

Browse source
This commit is contained in:
Alex Emmet 2026-03-09 22:29:48 +01:00
commit 657cdf50b1
2 changed files with 8 additions and 368 deletions
Download patch file Download diff file Expand all files Collapse all files

212
src/rho/connection.rs
View file

@ -60,25 +60,8 @@ impl GeneralConnection {
loop { loop {
let cv = match self.receiver.receive().await { let cv = match self.receiver.receive().await {
Ok(v) => { Ok(v) => v,
log_in!(
0,
PrintType::General,
"General connection received message type={:?} id={} identified={} challenged={}",
v.get_type(),
v.get_id(),
*self.identified.read().await,
*self.challenged.read().await
);
v
}
Err(e) => { Err(e) => {
log_err!(
0,
PrintType::General,
"General connection receive error before upgrade completion: {:?}",
e
);
break; break;
} }
}; };
@ -91,30 +74,12 @@ impl GeneralConnection {
if !*self.challenged.read().await { if !*self.challenged.read().await {
self.handle_challenge_response(cv).await; self.handle_challenge_response(cv).await;
if *self.challenged.read().await { if *self.challenged.read().await {
log_out!(
0,
PrintType::General,
"General connection challenge flow completed, handler will stop after immediate migration"
);
break; break;
} }
continue; continue;
} }
log_in!(
0,
PrintType::General,
"General connection ready to migrate for id={} kind={:?}",
*self.id.read().await,
*self.connection_kind.read().await
);
if self.migrate().await { if self.migrate().await {
log_out!(
0,
PrintType::General,
"General connection migration completed, handing over to specialized connection"
);
break; break;
} }
} }
@ -123,56 +88,23 @@ impl GeneralConnection {
} }
async fn handle_identification(self: &Arc<Self>, cv: CommunicationValue) { async fn handle_identification(self: &Arc<Self>, cv: CommunicationValue) {
if !cv.is_type(CommunicationType::identification) { if !cv.is_type(CommunicationType::identification) {
log_in!(
0,
PrintType::General,
"Ignoring pre-identification message type={:?} id={}",
cv.get_type(),
cv.get_id()
);
return; return;
} }
if let DataValue::Number(iota_id) = cv.get_data(DataTypes::iota_id) { if let DataValue::Number(iota_id) = cv.get_data(DataTypes::iota_id) {
log_in!(
*iota_id,
PrintType::Iota,
"Received Iota identification request message_id={}",
cv.get_id()
);
*self.id.write().await = *iota_id as u64; *self.id.write().await = *iota_id as u64;
*self.connection_kind.write().await = Some(ConnectionKind::Iota); *self.connection_kind.write().await = Some(ConnectionKind::Iota);
let get_pub_key_msg = CommunicationValue::new(CommunicationType::get_iota_data) let get_pub_key_msg = CommunicationValue::new(CommunicationType::get_iota_data)
.add_data(DataTypes::iota_id, DataValue::Number(*iota_id)); .add_data(DataTypes::iota_id, DataValue::Number(*iota_id));
log_out!(
*iota_id,
PrintType::Iota,
"Requesting Iota public key from Omega"
);
let response_cv = get_omega_connection() let response_cv = get_omega_connection()
.await_response(&get_pub_key_msg, Some(Duration::from_secs(20))) .await_response(&get_pub_key_msg, Some(Duration::from_secs(20)))
.await; .await;
let response_cv = match response_cv { let response_cv = match response_cv {
Ok(r) => { Ok(r) => r,
log_in!(
*iota_id,
PrintType::Iota,
"Received Iota public key response from Omega"
);
r
}
Err(e) => { Err(e) => {
log_err!(
*iota_id,
PrintType::Iota,
"Failed to load Iota public key from Omega: {:?}",
e
);
return; return;
} }
}; };
@ -185,11 +117,6 @@ impl GeneralConnection {
let pub_key = match load_public_key(base64_pub) { let pub_key = match load_public_key(base64_pub) {
Some(pk) => pk, Some(pk) => pk,
None => { None => {
log_err!(
*iota_id,
PrintType::Iota,
"Failed to decode Iota public key from Omega response"
);
return; return;
} }
}; };
@ -202,13 +129,6 @@ impl GeneralConnection {
.map(char::from) .map(char::from)
.collect(); .collect();
log_out!(
*iota_id,
PrintType::Iota,
"Generated challenge for Iota identification challenge_len={}",
challenge.len()
);
*self.challenge.write().await = challenge.clone(); *self.challenge.write().await = challenge.clone();
*self.identified.write().await = true; *self.identified.write().await = true;
@ -226,101 +146,32 @@ impl GeneralConnection {
) )
.add_data(DataTypes::challenge, DataValue::Str(encrypted_challenge)); .add_data(DataTypes::challenge, DataValue::Str(encrypted_challenge));
log_out!( let _ = self.sender.send(&response).await;
*iota_id,
PrintType::Iota,
"Sending encrypted identification challenge to Iota"
);
if let Err(e) = self.sender.send(&response).await {
log_err!(
*iota_id,
PrintType::Iota,
"Failed to send challenge to Iota: {:?}",
e
);
}
} else {
log_err!(
0,
PrintType::General,
"Identification message missing iota_id payload"
);
} }
} }
async fn handle_challenge_response(self: &Arc<Self>, cv: CommunicationValue) { async fn handle_challenge_response(self: &Arc<Self>, cv: CommunicationValue) {
let id = *self.id.read().await as i64; let id = *self.id.read().await as i64;
if !cv.is_type(CommunicationType::challenge_response) { if !cv.is_type(CommunicationType::challenge_response) {
log_in!(
id,
PrintType::Iota,
"Ignoring pre-challenge-completion message type={:?} id={}",
cv.get_type(),
cv.get_id()
);
return; return;
} }
if let DataValue::Str(response) = cv.get_data(DataTypes::challenge) { if let DataValue::Str(response) = cv.get_data(DataTypes::challenge) {
let expected = self.challenge.read().await.clone(); let expected = self.challenge.read().await.clone();
log_in!(
id,
PrintType::Iota,
"Received challenge response message_id={} response_len={} expected_len={}",
cv.get_id(),
response.len(),
expected.len()
);
if *response == expected { if *response == expected {
log_in!(
id,
PrintType::Iota,
"Challenge response validated successfully"
);
*self.challenged.write().await = true; *self.challenged.write().await = true;
let response = CommunicationValue::new(CommunicationType::identification_response) let response = CommunicationValue::new(CommunicationType::identification_response)
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_data(DataTypes::accepted, DataValue::Bool(true)); .add_data(DataTypes::accepted, DataValue::Bool(true));
log_out!( if let Err(_) = self.sender.send(&response).await {
id,
PrintType::Iota,
"Sending identification_response accepted=true"
);
if let Err(e) = self.sender.send(&response).await {
log_err!(
id,
PrintType::Iota,
"Failed to send identification_response: {:?}",
e
);
return; return;
} }
log_in!(
id,
PrintType::Iota,
"Immediately migrating upgraded connection after successful challenge validation"
);
if self.migrate().await { if self.migrate().await {
log_out!( log_out!(id, PrintType::Iota, "Immediate migration");
id,
PrintType::Iota,
"Immediate migration after challenge validation completed successfully"
);
} else {
log_err!(
id,
PrintType::Iota,
"Immediate migration after challenge validation failed"
);
} }
} else { } else {
log_err!( log_err!(
@ -344,87 +195,34 @@ impl GeneralConnection {
let kind = match *self.connection_kind.read().await { let kind = match *self.connection_kind.read().await {
Some(kind) => kind, Some(kind) => kind,
None => { None => {
log_err!(
0,
PrintType::General,
"Migration requested without a resolved connection kind"
);
return false; return false;
} }
}; };
let id = *self.id.read().await; let id = *self.id.read().await;
log_in!(
id as i64,
PrintType::General,
"Starting migration for kind={:?} id={}",
kind,
id
);
match kind { match kind {
ConnectionKind::Client => { ConnectionKind::Client => {
let client = ClientConnection::from_general(self.clone(), id).await; let client = ClientConnection::from_general(self.clone(), id).await;
client.start(); client.start();
log_out!(
id as i64,
PrintType::Client,
"Migrated general connection into ClientConnection"
);
} }
ConnectionKind::Iota => { ConnectionKind::Iota => {
let iota = IotaConnection::from_general(self.clone(), id).await; let iota = IotaConnection::from_general(self.clone(), id).await;
log_in!(
id as i64,
PrintType::Iota,
"Created upgraded IotaConnection from GeneralConnection"
);
let rho = Arc::new(RhoConnection::new(iota.clone(), Vec::new()).await); let rho = Arc::new(RhoConnection::new(iota.clone(), Vec::new()).await);
log_in!(
id as i64,
PrintType::Iota,
"Created RhoConnection for upgraded Iota connection"
);
iota.set_rho_connection(Arc::downgrade(&rho)).await; iota.set_rho_connection(Arc::downgrade(&rho)).await;
log_in!(
id as i64,
PrintType::Iota,
"Attached weak RhoConnection reference to IotaConnection"
);
rho_manager::add_rho(rho).await; rho_manager::add_rho(rho).await;
log_out!(
id as i64,
PrintType::Iota,
"Registered upgraded Iota connection in rho_manager"
);
iota.start(); iota.start();
log_out!(
id as i64,
PrintType::Iota,
"Started upgraded IotaConnection read loop"
);
} }
ConnectionKind::AnonymousClient => { ConnectionKind::AnonymousClient => {
let client = AnonymousClientConnection::from_general(self.clone(), id).await; let client = AnonymousClientConnection::from_general(self.clone(), id).await;
client.start(); client.start();
log_out!(
id as i64,
PrintType::Client,
"Migrated general connection into AnonymousClientConnection"
);
} }
ConnectionKind::Phi => { ConnectionKind::Phi => {
let iota = ClientConnection::from_general(self.clone(), id).await; let iota = ClientConnection::from_general(self.clone(), id).await;
iota.start(); iota.start();
log_out!(
id as i64,
PrintType::General,
"Migrated general connection into Phi/Client handler"
);
} }
} }
true true

164
src/rho/iota_connection.rs
View file

@ -56,41 +56,16 @@ impl IotaConnection {
pub fn start(self: Arc<Self>) { pub fn start(self: Arc<Self>) {
let self_clone = self.clone(); let self_clone = self.clone();
tokio::spawn(async move { tokio::spawn(async move {
log_in!(
self_clone.iota_id as i64,
PrintType::Iota,
"Upgraded IotaConnection read loop started"
);
loop { loop {
match self_clone.receiver.receive().await { match self_clone.receiver.receive().await {
Ok(cv) => { Ok(cv) => {
log_in!(
self_clone.iota_id as i64,
PrintType::Iota,
"Upgraded IotaConnection received message type={:?} id={}",
cv.get_type(),
cv.get_id()
);
self_clone.clone().handle_message(cv).await; self_clone.clone().handle_message(cv).await;
} }
Err(e) => { Err(_) => {
log_err!(
self_clone.iota_id as i64,
PrintType::Iota,
"Upgraded IotaConnection read loop stopped due to receive error: {:?}",
e
);
break; break;
} }
} }
} }
log_out!(
self_clone.iota_id as i64,
PrintType::Iota,
"Upgraded IotaConnection read loop exited"
);
}); });
} }
@ -150,23 +125,8 @@ impl IotaConnection {
/// Handle incoming message from Iota /// Handle incoming message from Iota
pub async fn handle_message(self: Arc<Self>, cv: CommunicationValue) { pub async fn handle_message(self: Arc<Self>, cv: CommunicationValue) {
log_in!(
self.iota_id as i64,
PrintType::Iota,
"Handling upgraded Iota message type={:?} id={} sender={} receiver={}",
cv.get_type(),
cv.get_id(),
cv.get_sender(),
cv.get_receiver()
);
// Handle ping // Handle ping
if cv.is_type(CommunicationType::ping) || cv.is_type(CommunicationType::pong) { if cv.is_type(CommunicationType::ping) || cv.is_type(CommunicationType::pong) {
log_in!(
self.iota_id as i64,
PrintType::Iota,
"Routing upgraded Iota message to ping handler"
);
self.handle_ping(cv).await; self.handle_ping(cv).await;
return; return;
} }
@ -175,11 +135,6 @@ impl IotaConnection {
// Handle GET_CHATS // Handle GET_CHATS
if cv.is_type(CommunicationType::get_chats) { if cv.is_type(CommunicationType::get_chats) {
log_in!(
self.iota_id as i64,
PrintType::Iota,
"Routing upgraded Iota message to get_chats handler"
);
self.handle_get_chats(cv).await; self.handle_get_chats(cv).await;
return; return;
} }
@ -190,12 +145,6 @@ impl IotaConnection {
|| cv.is_type(CommunicationType::message_other_iota) || cv.is_type(CommunicationType::message_other_iota)
|| cv.is_type(CommunicationType::send_chat) || cv.is_type(CommunicationType::send_chat)
{ {
log_in!(
self.iota_id as i64,
PrintType::Iota,
"Routing upgraded Iota message to forward_message handler receiver_id={}",
receiver_id
);
self.handle_forward_message(cv).await; self.handle_forward_message(cv).await;
return; return;
} }
@ -209,22 +158,10 @@ impl IotaConnection {
|| cv.is_type(CommunicationType::delete_iota) || cv.is_type(CommunicationType::delete_iota)
{ {
let sender = self.get_iota_id().await; let sender = self.get_iota_id().await;
log_in!(
self.iota_id as i64,
PrintType::Iota,
"Routing upgraded Iota message to Omega forwarder with sender={}",
sender
);
self.handle_omega_forward(cv.with_sender(sender as u64)) self.handle_omega_forward(cv.with_sender(sender as u64))
.await; .await;
return; return;
} }
// Forward to client
log_in!(
self.iota_id as i64,
PrintType::Iota,
"Routing upgraded Iota message to client forwarder"
);
self.forward_to_client(cv).await; self.forward_to_client(cv).await;
} }
@ -240,45 +177,16 @@ impl IotaConnection {
async fn handle_omega_forward(self: Arc<Self>, cv: CommunicationValue) { async fn handle_omega_forward(self: Arc<Self>, cv: CommunicationValue) {
let iota_for_closure = self.clone(); let iota_for_closure = self.clone();
tokio::spawn(async move { tokio::spawn(async move {
log_out!(
self.iota_id as i64,
PrintType::Iota,
"Forwarding upgraded Iota message to Omega type={:?} id={}",
cv.get_type(),
cv.get_id()
);
let response_cv = get_omega_connection() let response_cv = get_omega_connection()
.await_response(&cv.with_sender(self.iota_id), Some(Duration::from_secs(20))) .await_response(&cv.with_sender(self.iota_id), Some(Duration::from_secs(20)))
.await; .await;
if let Ok(response_cv) = response_cv { if let Ok(response_cv) = response_cv {
log_in!(
self.iota_id as i64,
PrintType::Iota,
"Received Omega response for upgraded Iota type={:?} id={}",
response_cv.get_type(),
response_cv.get_id()
);
iota_for_closure.send_message(&response_cv).await; iota_for_closure.send_message(&response_cv).await;
} else if let Err(e) = response_cv {
log_err!(
self.iota_id as i64,
PrintType::Iota,
"Omega forward failed for upgraded Iota connection: {}",
e
);
} }
}); });
} }
/// Handle ping message /// Handle ping message
async fn handle_ping(&self, cv: CommunicationValue) { async fn handle_ping(&self, cv: CommunicationValue) {
log_in!(
self.iota_id as i64,
PrintType::Iota,
"Handling ping/pong for upgraded Iota connection message_id={}",
cv.get_id()
);
if let DataValue::Number(last_ping) = cv.get_data(DataTypes::last_ping) { if let DataValue::Number(last_ping) = cv.get_data(DataTypes::last_ping) {
if let Ok(ping_val) = last_ping.to_string().parse::<i64>() { if let Ok(ping_val) = last_ping.to_string().parse::<i64>() {
let mut ping_guard = self.ping.write().await; let mut ping_guard = self.ping.write().await;
@ -300,13 +208,6 @@ impl IotaConnection {
.with_id(cv.get_id()) .with_id(cv.get_id())
.add_data(DataTypes::ping_clients, DataValue::Container(pings)); .add_data(DataTypes::ping_clients, DataValue::Container(pings));
log_out!(
self.iota_id as i64,
PrintType::Iota,
"Sending pong from upgraded Iota connection message_id={}",
cv.get_id()
);
self.send_message(&response).await; self.send_message(&response).await;
} }
@ -315,43 +216,16 @@ impl IotaConnection {
let receiver_id = cv.get_receiver(); let receiver_id = cv.get_receiver();
let sender_id = cv.get_sender(); let sender_id = cv.get_sender();
log_in!(
self.iota_id as i64,
PrintType::Iota,
"Handling cross-routing message sender={} receiver={} type={:?}",
sender_id,
receiver_id,
cv.get_type()
);
if self.get_user_ids().await.contains(&(sender_id as u64)) { if self.get_user_ids().await.contains(&(sender_id as u64)) {
if let Some(target_rho) = rho_manager::get_rho_con_for_user(receiver_id as i64).await { if let Some(target_rho) = rho_manager::get_rho_con_for_user(receiver_id as i64).await {
log_out!(
self.iota_id as i64,
PrintType::Iota,
"Forwarding upgraded Iota message to target rho receiver={}",
receiver_id
);
target_rho.message_to_iota(cv).await; target_rho.message_to_iota(cv).await;
} else { } else {
log_err!(
self.iota_id as i64,
PrintType::Iota,
"No target rho found for receiver={}",
receiver_id
);
let error = CommunicationValue::new(CommunicationType::error_no_iota) let error = CommunicationValue::new(CommunicationType::error_no_iota)
.with_id(cv.get_id()) .with_id(cv.get_id())
.with_sender(cv.get_sender()); .with_sender(cv.get_sender());
self.send_message(&error).await; self.send_message(&error).await;
} }
} else { } else {
log_err!(
self.iota_id as i64,
PrintType::Iota,
"Unauthorized sender attempted cross-routing sender={}",
sender_id
);
self.send_message( self.send_message(
&CommunicationValue::new(CommunicationType::error_invalid_user_id).add_data( &CommunicationValue::new(CommunicationType::error_invalid_user_id).add_data(
DataTypes::error_type, DataTypes::error_type,
@ -469,39 +343,14 @@ impl IotaConnection {
async fn forward_to_client(&self, cv: CommunicationValue) { async fn forward_to_client(&self, cv: CommunicationValue) {
if let Some(rho_conn) = self.get_rho_connection().await { if let Some(rho_conn) = self.get_rho_connection().await {
let updated_cv = cv.with_sender(self.get_iota_id().await); let updated_cv = cv.with_sender(self.get_iota_id().await);
log_out!(
self.get_iota_id().await as i64,
PrintType::Iota,
"Forwarding upgraded Iota message to client type={:?} id={} receiver={}",
updated_cv.get_type(),
updated_cv.get_id(),
updated_cv.get_receiver()
);
rho_conn.message_to_client(updated_cv).await; rho_conn.message_to_client(updated_cv).await;
} else { } else {
log_err!(
self.get_iota_id().await as i64,
PrintType::General,
"Failed to forward message to client because rho connection is missing"
);
} }
} }
pub async fn handle_close(&self) { pub async fn handle_close(&self) {
log_out!(
self.get_iota_id().await as i64,
PrintType::Iota,
"Handling upgraded Iota connection close"
);
if let Some(rho_conn) = self.get_rho_connection().await { if let Some(rho_conn) = self.get_rho_connection().await {
rho_conn.close_iota_connection().await; rho_conn.close_iota_connection().await;
} else {
log_err!(
self.get_iota_id().await as i64,
PrintType::Iota,
"No rho connection available during upgraded Iota close"
);
} }
} }
@ -516,17 +365,10 @@ impl IotaConnection {
let task_tx = tx.clone(); let task_tx = tx.clone();
self.waiting_tasks.insert( self.waiting_tasks.insert(
msg_id, msg_id,
Box::new(move |io, response_cv| { Box::new(move |_, response_cv| {
let inner_tx = task_tx.clone(); let inner_tx = task_tx.clone();
tokio::spawn(async move { tokio::spawn(async move {
if let Err(e) = inner_tx.send(response_cv).await { let _ = inner_tx.send(response_cv).await;
log_err!(
io.get_iota_id().await as i64,
PrintType::Iota,
"Failed to send response back to awaiter: {}",
e
);
}
}); });
true true
}), }),