From f2770fb98f90b909d81b7e246abb1fbcf7e6132f Mon Sep 17 00:00:00 2001 From: Rasensprenger Date: Sun, 30 Aug 2026 21:01:15 +0300 Subject: [PATCH 1/3] Update Rust crate rand to 0.10 --- Cargo.lock | 43 +---------------------------------------- communities/Cargo.toml | 2 +- iota-storage/Cargo.toml | 2 +- 3 files changed, 3 insertions(+), 44 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e506091..92d3dbe 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2247,7 +2247,7 @@ dependencies = [ "json", "once_cell", "r2d2", - "rand 0.8.7", + "rand 0.10.2", "rusqlite", "serde", "serde_yaml", @@ -3416,15 +3416,6 @@ version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "efca4c95a19a79d1c98f791f10aebd5c1363b473244630bb7dbde1dc98455a24" -[[package]] -name = "ppv-lite86" -version = "0.2.21" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" -dependencies = [ - "zerocopy", -] - [[package]] name = "proc-macro2" version = "1.0.107" @@ -3532,8 +3523,6 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "22f6172bdec972074665ed81ed53b71da00bfc44b65a753cfde883ec4c702a1a" dependencies = [ - "libc", - "rand_chacha", "rand_core 0.6.4", ] @@ -3548,16 +3537,6 @@ dependencies = [ "rand_core 0.10.1", ] -[[package]] -name = "rand_chacha" -version = "0.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" -dependencies = [ - "ppv-lite86", - "rand_core 0.6.4", -] - [[package]] name = "rand_core" version = "0.6.4" @@ -5457,26 +5436,6 @@ dependencies = [ "synstructure", ] -[[package]] -name = "zerocopy" -version = "0.8.56" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "556764e583adb45a9f8d413c2a147fa7e8d821e48e12b14fd560b607998b75eb" -dependencies = [ - "zerocopy-derive", -] - -[[package]] -name = "zerocopy-derive" -version = "0.8.56" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f2ab42fc20575779bd240faa45f94a74256f755c0fa9e89f0ede20d91d0cdfc1" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.119", -] - [[package]] name = "zerofrom" version = "0.1.8" diff --git a/communities/Cargo.toml b/communities/Cargo.toml index 36ef42e..0b3ad25 100644 --- a/communities/Cargo.toml +++ b/communities/Cargo.toml @@ -8,7 +8,7 @@ mtp = { git = "https://git.methanium.net/Methanium/mtp.git" } iota-util = { path = "../iota-util" } futures = "*" -rand = "0.8" +rand = "0.10" rand_core = { version = "0.6", features = ["getrandom", "std"] } sha2 = "0.11.0" tokio = { version = "1.50.0", features = ["full"] } diff --git a/iota-storage/Cargo.toml b/iota-storage/Cargo.toml index 9f91c52..5541ef4 100644 --- a/iota-storage/Cargo.toml +++ b/iota-storage/Cargo.toml @@ -15,6 +15,6 @@ r2d2 = "0.8" serde = { version = "1", features = ["derive"] } serde_yaml = "0.9" thiserror = "2" -rand = "0.8" +rand = "0.10" rusqlite = "0.40.0" tokio = { version = "1.50.0", features = ["full"] } From ec3f5e6a6eedb40a7f4d91aa4f987c00bf601cea Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Sun, 30 Aug 2026 21:28:17 +0200 Subject: [PATCH 2/3] [Fix] Durability --- client/src/client_connection.rs | 5 - iota-connection/src/message_handlers.rs | 54 ++++---- iota-storage/src/users/contact.rs | 15 +-- iota-storage/src/users/user_profile.rs | 28 ++++- iota-storage/src/util/chat_files.rs | 130 ++++++++++++++------ iota-storage/src/util/chats_util.rs | 72 ++++++----- iota-storage/src/util/db.rs | 58 ++++++++- omikron-connector/src/omikron_connection.rs | 50 +++++++- omikron-connector/src/user_ops.rs | 25 +++- 9 files changed, 318 insertions(+), 119 deletions(-) diff --git a/client/src/client_connection.rs b/client/src/client_connection.rs index f52f6f1..a43c308 100644 --- a/client/src/client_connection.rs +++ b/client/src/client_connection.rs @@ -226,11 +226,6 @@ impl ClientConnection { // Direct messages // // ************************************************ // - if cv.is_type(CommunicationType::MessageState) { - message_handlers::handle_message_state(&cv); - return; - } - if cv.is_type(CommunicationType::MessageEdit) { self.send_message(&message_handlers::handle_message_edit(&cv)) .await; diff --git a/iota-connection/src/message_handlers.rs b/iota-connection/src/message_handlers.rs index f912259..7da9a1d 100644 --- a/iota-connection/src/message_handlers.rs +++ b/iota-connection/src/message_handlers.rs @@ -72,7 +72,6 @@ fn add_conversation_for_user( if let Some(name) = name { contact.user_name = Some(name.to_string()); } - contact.set_last_message_at(now_millis_i64()); mod_user(user_id, &contact) } @@ -228,7 +227,11 @@ pub fn apply_verified_relay_content( reply_to, origin_iota_received_at: sent_by_self.then_some(accepted_at), destination_iota_received_at: (!sent_by_self).then_some(accepted_at), - initial_state: MessageState::Sent, + initial_state: if sent_by_self { + MessageState::Sending + } else { + MessageState::Sent + }, }) .map_err(|error| error.to_string())?; Ok(()) @@ -430,6 +433,12 @@ fn stored_message_fields( DataValue::Str(relay_message_id.clone()), )); } + if let Some(relay_signer_id) = message.relay_signer_id { + fields.push(( + DataType::SenderId, + DataValue::SignedNumber(relay_signer_id.into()), + )); + } for (data_type, timestamp) in [ (DataType::AuthoredAt, message.authored_at), ( @@ -452,6 +461,15 @@ fn stored_message_fields( fields.push((data_type, DataValue::SignedNumber(timestamp.into()))); } } + if let Some(failed_at) = message.delivery_failed_at { + fields.push(( + DataType::UpdatedAt, + DataValue::SignedNumber(failed_at.into()), + )); + } + if let Some(failure) = &message.delivery_failure { + fields.push((DataType::ErrorType, DataValue::Str(failure.clone()))); + } if message.edited { fields.push((DataType::Edited, DataValue::Bool(true))); } @@ -644,6 +662,12 @@ fn contact_value( if let Some(name) = &contact.user_name { fields.push((DataType::Username, DataValue::Str(name.clone()))); } + if contact.created_at > 0 { + fields.push(( + DataType::CreatedAt, + DataValue::SignedNumber(contact.created_at.into()), + )); + } if let Some(last_message_at) = contact.last_message_at { fields.push(( DataType::LastMessageAt, @@ -918,26 +942,6 @@ pub fn handle_client_state_ack(cv: &CommunicationValue) -> CommunicationValue { ) } -pub fn handle_message_state(cv: &CommunicationValue) { - let sender_id = match required_sender_id(cv) { - Ok(sender_id) => sender_id, - Err(_) => return, - }; - let receiver_id = match data_i64(cv, DataType::ChatPartnerId) { - Some(id) if id > 0 => id, - _ => return, - }; - - let timestamp_i64 = data_i64(cv, DataType::SendTime).unwrap_or_else(now_millis_i64); - - let _ = chat_files::change_message_state( - timestamp_i64, - receiver_id, - sender_id, - MessageState::from_str(cv.get_data(DataType::MessageState).as_str().unwrap_or("")), - ); -} - pub fn handle_messages_get(cv: &CommunicationValue) -> CommunicationValue { let my_id = match cv.require_sender() { Ok(my_id) => my_id, @@ -1026,6 +1030,12 @@ pub fn handle_get_chats(cv: &CommunicationValue) -> CommunicationValue { if let Some(name) = user.user_name { container.push((DataType::Username, DataValue::Str(name))); } + if user.created_at > 0 { + container.push(( + DataType::CreatedAt, + DataValue::SignedNumber(user.created_at.into()), + )); + } if let Some(ts) = user.last_message_at { container.push((DataType::LastMessageAt, DataValue::SignedNumber(ts as i128))); } diff --git a/iota-storage/src/users/contact.rs b/iota-storage/src/users/contact.rs index 50d224d..4d2ca43 100644 --- a/iota-storage/src/users/contact.rs +++ b/iota-storage/src/users/contact.rs @@ -1,31 +1,32 @@ -use std::time::{SystemTime, UNIX_EPOCH}; - #[derive(Debug, Clone)] pub struct Contact { pub user_id: i64, pub user_name: Option, + pub created_at: i64, pub last_message_at: Option, } impl Default for Contact { fn default() -> Self { - let now = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap() - .as_millis() as i64; Contact { user_id: 0, user_name: None, - last_message_at: Some(now), + created_at: 0, + last_message_at: None, } } } impl Contact { pub fn new(user_id: i64) -> Self { + let created_at = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i64; Contact { user_id: user_id, user_name: None, + created_at, last_message_at: None, } } diff --git a/iota-storage/src/users/user_profile.rs b/iota-storage/src/users/user_profile.rs index 5f2abd5..f1aa2a3 100644 --- a/iota-storage/src/users/user_profile.rs +++ b/iota-storage/src/users/user_profile.rs @@ -27,6 +27,29 @@ impl UserProfile { public_key: String, private_key_hash: String, reset_token: String, + ) -> Self { + Self::new_with_created_at( + user_id, + username, + display_name, + public_key, + private_key_hash, + reset_token, + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i64, + ) + } + + pub fn new_with_created_at( + user_id: i64, + username: String, + display_name: Option, + public_key: String, + private_key_hash: String, + reset_token: String, + created_at: i64, ) -> Self { Self { user_id, @@ -34,10 +57,7 @@ impl UserProfile { display_name, public_key, private_key_hash, - created_at: SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap() - .as_millis() as i64, + created_at, reset_token, trusted_apps: std::collections::HashMap::new(), } diff --git a/iota-storage/src/util/chat_files.rs b/iota-storage/src/util/chat_files.rs index 94aefb7..051557b 100644 --- a/iota-storage/src/util/chat_files.rs +++ b/iota-storage/src/util/chat_files.rs @@ -60,6 +60,8 @@ pub struct StoredMessage { pub client_received_recorded_at: Option, pub read_at: Option, pub read_recorded_at: Option, + pub delivery_failed_at: Option, + pub delivery_failure: Option, pub content: String, pub edited: bool, pub sent_by_self: bool, @@ -483,7 +485,7 @@ pub fn add_message(message: NewMessage<'_>) -> Result { destination_iota_received_at, initial_state, } = message; - let msg_id = db::with_db(|conn| { + db::with_db(|conn| { let tx = conn.unchecked_transaction()?; tx.execute( r#" @@ -517,14 +519,16 @@ pub fn add_message(message: NewMessage<'_>) -> Result { msg_id, Operation::Upsert, )?; + let mut contact = crate::users::contact::Contact::new(external_user); + contact.set_last_message_at( + destination_iota_received_at + .or(origin_iota_received_at) + .unwrap_or(authored_at), + ); + crate::util::chats_util::upsert_contact(&tx, storage_owner, &contact)?; tx.commit()?; Ok(msg_id) - })?; - - let mut contact = crate::users::contact::Contact::new(external_user); - contact.set_last_message_at(send_time); - crate::util::chats_util::mod_user(storage_owner, &contact)?; - Ok(msg_id) + }) } pub fn change_message_state_by_relay_id( @@ -579,11 +583,11 @@ pub fn record_message_receipt( }; db::with_db(|conn| { let tx = conn.unchecked_transaction()?; - let Some((message_id, external_user)) = tx + let Some((message_id, external_user, authored_at)) = tx .query_row( - "SELECT id, external_user FROM messages WHERE storage_owner = ?1 AND relay_signer_id = ?2 AND relay_message_id = ?3", + "SELECT id, external_user, authored_at FROM messages WHERE storage_owner = ?1 AND relay_signer_id = ?2 AND relay_message_id = ?3", params![storage_owner, target_signer_id, target_message_id], - |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)), + |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?, row.get::<_, Option>(2)?)), ) .optional()? else { @@ -594,6 +598,13 @@ pub fn record_message_receipt( "message receipt signer is not the chat partner".into(), )); } + if event_at > recorded_at.saturating_add(5 * 60 * 1000) + || authored_at.is_some_and(|authored_at| event_at < authored_at) + { + return Err(StorageError::Other( + "message receipt event time is outside the accepted clock range".into(), + )); + } tx.execute( "INSERT OR IGNORE INTO message_receipts (storage_owner, target_signer_id, target_message_id, receipt_signer_id, receipt_message_id, receipt_type, event_at, recorded_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", params![storage_owner, target_signer_id, target_message_id, receipt_signer_id, receipt_message_id, receipt_type, event_at, recorded_at], @@ -646,7 +657,7 @@ pub fn record_destination_iota_received( return Ok(()); }; tx.execute( - "UPDATE messages SET destination_iota_received_at = COALESCE(destination_iota_received_at, ?1), message_state = CASE WHEN message_state = 'sending' THEN 'sent' ELSE message_state END WHERE id = ?2", + "UPDATE messages SET destination_iota_received_at = COALESCE(destination_iota_received_at, ?1), delivery_failed_at = NULL, delivery_failure = NULL, message_state = CASE WHEN message_state = 'sending' THEN 'sent' ELSE message_state END WHERE id = ?2", params![accepted_at, message_id], )?; sync::record_event( @@ -661,6 +672,41 @@ pub fn record_destination_iota_received( }) } +pub fn record_delivery_failure( + storage_owner: i64, + relay_signer_id: i64, + relay_message_id: &str, + failure: &str, + failed_at: i64, +) -> Result<(), StorageError> { + db::with_db(|conn| { + let tx = conn.unchecked_transaction()?; + let Some(message_id) = tx + .query_row( + "SELECT id FROM messages WHERE storage_owner = ?1 AND relay_signer_id = ?2 AND relay_message_id = ?3", + params![storage_owner, relay_signer_id, relay_message_id], + |row| row.get::<_, i64>(0), + ) + .optional()? + else { + return Ok(()); + }; + tx.execute( + "UPDATE messages SET delivery_failed_at = ?1, delivery_failure = ?2 WHERE id = ?3 AND destination_iota_received_at IS NULL", + params![failed_at, failure, message_id], + )?; + sync::record_event( + &tx, + storage_owner, + EntityType::Message, + message_id, + Operation::Upsert, + )?; + tx.commit()?; + Ok(()) + }) +} + pub fn change_message_state( timestamp: i64, storage_owner: i64, @@ -783,11 +829,11 @@ pub fn get_messages( SELECT id, relay_signer_id, relay_message_id, message_time, authored_at, origin_iota_received_at, destination_iota_received_at, client_received_at, client_received_recorded_at, read_at, - read_recorded_at, content, sent_by_self, message_state, height, + read_recorded_at, delivery_failed_at, delivery_failure, content, sent_by_self, message_state, height, reply_to, edited_count FROM messages WHERE storage_owner = ?1 AND external_user = ?2 AND deleted_by_external = 0 - ORDER BY message_time DESC, id DESC + ORDER BY COALESCE(destination_iota_received_at, origin_iota_received_at, authored_at, id) DESC, id DESC LIMIT ?3 OFFSET ?4 "#, )?; @@ -808,12 +854,14 @@ pub fn get_messages( client_received_recorded_at: row.get(8)?, read_at: row.get(9)?, read_recorded_at: row.get(10)?, - content: row.get(11)?, - sent_by_self: row.get::<_, i64>(12)? != 0, - message_state: row.get(13)?, - height: row.get(14).unwrap_or(0), - reply_to: row.get(15).ok().flatten(), - edited: row.get::<_, i64>(16).unwrap_or(0) > 0, + delivery_failed_at: row.get(11)?, + delivery_failure: row.get(12)?, + content: row.get(13)?, + sent_by_self: row.get::<_, i64>(14)? != 0, + message_state: row.get(15)?, + height: row.get(16).unwrap_or(0), + reply_to: row.get(17).ok().flatten(), + edited: row.get::<_, i64>(18).unwrap_or(0) > 0, reactions: Vec::new(), }) }, @@ -854,7 +902,7 @@ pub fn get_message( SELECT id, relay_signer_id, relay_message_id, message_time, authored_at, origin_iota_received_at, destination_iota_received_at, client_received_at, client_received_recorded_at, read_at, - read_recorded_at, content, sent_by_self, message_state, height, + read_recorded_at, delivery_failed_at, delivery_failure, content, sent_by_self, message_state, height, reply_to, edited_count, external_user FROM messages WHERE storage_owner = ?1 @@ -878,13 +926,15 @@ pub fn get_message( client_received_recorded_at: row.get(8)?, read_at: row.get(9)?, read_recorded_at: row.get(10)?, - content: row.get(11)?, - sent_by_self: row.get::<_, i64>(12)? != 0, - message_state: row.get(13)?, - height: row.get(14).unwrap_or(0), - reply_to: row.get(15).ok().flatten(), - edited: row.get::<_, i64>(16).unwrap_or(0) > 0, - external_user: row.get(17)?, + delivery_failed_at: row.get(11)?, + delivery_failure: row.get(12)?, + content: row.get(13)?, + sent_by_self: row.get::<_, i64>(14)? != 0, + message_state: row.get(15)?, + height: row.get(16).unwrap_or(0), + reply_to: row.get(17).ok().flatten(), + edited: row.get::<_, i64>(18).unwrap_or(0) > 0, + external_user: row.get(19)?, reactions: Vec::new(), }) })?; @@ -928,14 +978,16 @@ pub fn get_message_with_offset( AND external_user = ?2 AND deleted_by_external = 0 AND ( - message_time > ?3 - OR (message_time = ?3 AND id > ?4) + COALESCE(destination_iota_received_at, origin_iota_received_at, authored_at, id) > + COALESCE(?3, ?4) + OR (COALESCE(destination_iota_received_at, origin_iota_received_at, authored_at, id) = + COALESCE(?3, ?4) AND id > ?4) ) "#, params![ storage_owner, external_user, - message.message_time, + message.destination_iota_received_at.or(message.origin_iota_received_at).or(message.authored_at), message.id ], |row| row.get(0), @@ -953,9 +1005,9 @@ pub fn get_messages_by_ids(storage_owner: i64, ids: &[i64]) -> Vec Vec(12)? != 0, - message_state: row.get(13)?, - height: row.get(14).unwrap_or(0), - reply_to: row.get(15).ok().flatten(), - edited: row.get::<_, i64>(16).unwrap_or(0) > 0, + delivery_failed_at: row.get(11)?, + delivery_failure: row.get(12)?, + content: row.get(13)?, + sent_by_self: row.get::<_, i64>(14)? != 0, + message_state: row.get(15)?, + height: row.get(16).unwrap_or(0), + reply_to: row.get(17).ok().flatten(), + edited: row.get::<_, i64>(18).unwrap_or(0) > 0, reactions: Vec::new(), }) })?; diff --git a/iota-storage/src/util/chats_util.rs b/iota-storage/src/util/chats_util.rs index 1fd3cb0..3eb4043 100644 --- a/iota-storage/src/util/chats_util.rs +++ b/iota-storage/src/util/chats_util.rs @@ -4,6 +4,42 @@ use crate::util::db; use crate::util::sync::{self, EntityType, Operation}; use rusqlite::params; +pub(crate) fn upsert_contact( + tx: &rusqlite::Transaction<'_>, + storage_owner: i64, + contact: &Contact, +) -> Result<(), StorageError> { + tx.execute( + r#" + INSERT INTO contacts (storage_owner, user_id, user_name, created_at, last_message_at) + VALUES (?1, ?2, ?3, ?4, ?5) + ON CONFLICT(storage_owner, user_id) DO UPDATE SET + user_name = COALESCE(excluded.user_name, contacts.user_name), + created_at = MIN(contacts.created_at, excluded.created_at), + last_message_at = CASE + WHEN excluded.last_message_at IS NULL THEN contacts.last_message_at + WHEN contacts.last_message_at IS NULL THEN excluded.last_message_at + ELSE MAX(contacts.last_message_at, excluded.last_message_at) + END + "#, + params![ + storage_owner, + contact.user_id, + contact.user_name, + contact.created_at, + contact.last_message_at, + ], + )?; + sync::record_event( + tx, + storage_owner, + EntityType::Contact, + contact.user_id, + Operation::Upsert, + )?; + Ok(()) +} + pub fn has_user(storage_owner: i64, user_id: i64) -> Result { db::with_db(|conn| { Ok(conn.query_row( @@ -15,31 +51,7 @@ pub fn has_user(storage_owner: i64, user_id: i64) -> Result } pub fn mod_user(storage_owner: i64, contact: &Contact) -> Result<(), StorageError> { - db::with_immediate_transaction(|tx| { - tx.execute( - r#" - INSERT INTO contacts (storage_owner, user_id, user_name, last_message_at) - VALUES (?1, ?2, ?3, ?4) - ON CONFLICT(storage_owner, user_id) DO UPDATE SET - user_name = excluded.user_name, - last_message_at = excluded.last_message_at - "#, - params![ - storage_owner, - contact.user_id, - contact.user_name, - contact.last_message_at, - ], - )?; - sync::record_event( - tx, - storage_owner, - EntityType::Contact, - contact.user_id, - Operation::Upsert, - )?; - Ok(()) - }) + db::with_immediate_transaction(|tx| upsert_contact(tx, storage_owner, contact)) } pub fn get_users_by_ids(storage_owner: i64, ids: &[i64]) -> Result, StorageError> { @@ -57,7 +69,7 @@ pub fn get_user(storage_owner: i64, user_id: i64) -> Result, Sto db::with_db(|conn| { match conn.query_row( r#" - SELECT user_id, user_name, last_message_at + SELECT user_id, user_name, created_at, last_message_at FROM contacts WHERE storage_owner = ?1 AND user_id = ?2 LIMIT 1 @@ -67,7 +79,8 @@ pub fn get_user(storage_owner: i64, user_id: i64) -> Result, Sto Ok(Contact { user_id: r.get(0)?, user_name: r.get(1)?, - last_message_at: r.get(2)?, + created_at: r.get(2)?, + last_message_at: r.get(3)?, }) }, ) { @@ -82,7 +95,7 @@ pub fn get_users(storage_owner: i64) -> Result, StorageError> { db::with_db(|conn| { let mut stmt = conn.prepare( r#" - SELECT user_id, user_name, last_message_at + SELECT user_id, user_name, created_at, last_message_at FROM contacts WHERE storage_owner = ?1 ORDER BY @@ -96,7 +109,8 @@ pub fn get_users(storage_owner: i64) -> Result, StorageError> { Ok(Contact { user_id: r.get(0)?, user_name: r.get(1)?, - last_message_at: r.get(2)?, + created_at: r.get(2)?, + last_message_at: r.get(3)?, }) })?; diff --git a/iota-storage/src/util/db.rs b/iota-storage/src/util/db.rs index 3e69a6f..e8cbf6e 100644 --- a/iota-storage/src/util/db.rs +++ b/iota-storage/src/util/db.rs @@ -180,6 +180,7 @@ fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> { storage_owner INTEGER NOT NULL, user_id INTEGER NOT NULL, user_name TEXT, + created_at INTEGER NOT NULL, last_message_at INTEGER, UNIQUE(storage_owner, user_id) ); @@ -527,6 +528,57 @@ fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> { )?; } + if current_version < 15 { + let messages_exist: bool = conn.query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'messages')", + [], + |row| row.get(0), + )?; + if messages_exist { + conn.execute_batch( + r#" + DROP INDEX IF EXISTS idx_messages_history; + CREATE INDEX IF NOT EXISTS idx_messages_history_accepted + ON messages ( + storage_owner, + external_user, + deleted_by_external, + destination_iota_received_at DESC, + origin_iota_received_at DESC, + authored_at DESC, + id DESC + ); + "#, + )?; + } + conn.pragma_update(None, "user_version", 15)?; + } + + if current_version < 16 { + let messages_exist: bool = conn.query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'messages')", + [], + |row| row.get(0), + )?; + if messages_exist { + add_column_if_missing(conn, "delivery_failed_at", "delivery_failed_at INTEGER")?; + add_column_if_missing(conn, "delivery_failure", "delivery_failure TEXT")?; + } + let contacts_exist: bool = conn.query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'contacts')", + [], + |row| row.get(0), + )?; + if contacts_exist { + add_table_column_if_missing(conn, "contacts", "created_at", "created_at INTEGER")?; + conn.execute( + "UPDATE contacts SET created_at = COALESCE(created_at, last_message_at, 0)", + [], + )?; + } + conn.pragma_update(None, "user_version", 16)?; + } + Ok(()) } @@ -596,7 +648,7 @@ mod tests { run_migrations_on_connection(&conn)?; let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?; - assert_eq!(version, 13); + assert_eq!(version, 16); for column in ["height", "reply_to", "edited_count", "deleted_by_external"] { let mut statement = conn.prepare("SELECT 1 FROM pragma_table_info('messages') WHERE name = ?1")?; @@ -615,7 +667,7 @@ mod tests { run_migrations_on_connection(&conn)?; run_migrations_on_connection(&conn)?; let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?; - assert_eq!(version, 13); + assert_eq!(version, 16); for table in [ "sync_heads", "sync_events", @@ -650,7 +702,7 @@ mod tests { run_migrations_on_connection(&conn)?; let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?; - assert_eq!(version, 13); + assert_eq!(version, 16); for column in [ "id", "user_id", diff --git a/omikron-connector/src/omikron_connection.rs b/omikron-connector/src/omikron_connection.rs index 5513efa..adb18ee 100644 --- a/omikron-connector/src/omikron_connection.rs +++ b/omikron-connector/src/omikron_connection.rs @@ -40,6 +40,21 @@ const IOTA_KEYRING_PATH: &str = "iota.mk"; static IDENTITY_PATH: std::sync::OnceLock = std::sync::OnceLock::new(); static OMIKRON_PUBLIC_KEY_PATH: std::sync::OnceLock = std::sync::OnceLock::new(); +fn record_origin_delivery_failure(signer_id: u64, relay_message_id: &str, failure: &str) { + let Ok(storage_owner) = i64::try_from(signer_id) else { + return; + }; + if let Err(error) = chat_files::record_delivery_failure( + storage_owner, + storage_owner, + relay_message_id, + failure, + now_millis_i64(), + ) { + log!("Relay delivery failure storage failed: {error}"); + } +} + /* * Keeps identity and pinned Omikron key files independent from the process * working directory, so restarts use the same trusted material. @@ -1136,6 +1151,11 @@ impl OmikronConnection { Ok(destination_iota) => destination_iota, Err(error) => { log!("Relay origin route lookup failed: {}", error); + record_origin_delivery_failure( + verified.context.signer_id, + &verified.context.message_id, + "destination_iota_not_found", + ); self.send_relay_response(frame.id(), CommunicationType::ErrorNoIota) .await; return; @@ -1145,6 +1165,11 @@ impl OmikronConnection { Ok(value) => value, Err(error) => { log!("Relay origin forwarding validation failed: {}", error); + record_origin_delivery_failure( + verified.context.signer_id, + &verified.context.message_id, + "forwarding_validation_failed", + ); self.send_relay_response(frame.id(), CommunicationType::ErrorInvalidData) .await; return; @@ -1154,6 +1179,11 @@ impl OmikronConnection { Ok(bytes) => bytes, Err(error) => { log!("Relay origin retry could not be serialized: {}", error); + record_origin_delivery_failure( + verified.context.signer_id, + &verified.context.message_id, + "serialization_failed", + ); self.send_relay_response(frame.id(), CommunicationType::ErrorInternal) .await; return; @@ -1167,6 +1197,11 @@ impl OmikronConnection { &type_map_version, ) { log!("Relay origin retry queue failed: {}", error); + record_origin_delivery_failure( + verified.context.signer_id, + &verified.context.message_id, + "queue_failed", + ); self.send_relay_response(frame.id(), CommunicationType::ErrorInternal) .await; return; @@ -1242,6 +1277,11 @@ impl OmikronConnection { } Ok(response) => { log!("Relay origin route returned {}", response.get_type()); + record_origin_delivery_failure( + verified.context.signer_id, + &verified.context.message_id, + "destination_rejected", + ); self.send_relay_response( frame.id(), response @@ -1252,6 +1292,11 @@ impl OmikronConnection { } Err(error) => { log!("Relay origin forwarding failed: {}", error); + record_origin_delivery_failure( + verified.context.signer_id, + &verified.context.message_id, + "destination_unreachable", + ); self.send_relay_response(frame.id(), CommunicationType::ErrorInternal) .await; } @@ -1521,7 +1566,6 @@ impl OmikronConnection { dispatch!(DeleteApp, handle_delete_app); dispatch!(ClientConnected, handle_client_connected); dispatch!(ClientStateAck, handle_client_state_ack); - dispatch!(MessageState, handle_message_state); dispatch!(MessageEdit, handle_message_edit); dispatch!(MessageEditLive, handle_message_edit_live); dispatch!(MessageReactionAdd, handle_message_reaction_add); @@ -1771,10 +1815,6 @@ impl OmikronConnection { .await; } - async fn handle_message_state(self: Arc, cv: &CommunicationValue) { - message_handlers::handle_message_state(cv); - } - fn mutation_live_message( ty: CommunicationType, request: &CommunicationValue, diff --git a/omikron-connector/src/user_ops.rs b/omikron-connector/src/user_ops.rs index b140358..5c33416 100644 --- a/omikron-connector/src/user_ops.rs +++ b/omikron-connector/src/user_ops.rs @@ -84,7 +84,7 @@ fn sign_lifecycle_payload( async fn inspect_credential_account( connection: &dyn OmikronClient, credential: &TuCredential, -) -> Result<(String, String), LifecycleUserError> { +) -> Result<(String, String, i64), LifecycleUserError> { if credential.omega_host != omega_discovery::omega_host() { return Err(LifecycleUserError::OmegaHostMismatch); } @@ -108,10 +108,16 @@ async fn inspect_credential_account( .as_str() .map(str::to_owned) .ok_or(LifecycleUserError::RemoteRejected)?; + let created_at = response + .get_data(DataType::CreatedAt) + .as_signed_number() + .and_then(|value| i64::try_from(value).ok()) + .filter(|value| *value > 0) + .ok_or(LifecycleUserError::RemoteRejected)?; if public_key != public_key_bundle_to_base64(&credential.public_key_bundle()) { return Err(LifecycleUserError::RemoteRejected); } - Ok((username, public_key)) + Ok((username, public_key, created_at)) } async fn credential_proof( @@ -168,17 +174,17 @@ pub async fn attach_user_from_tu( ) -> Result { let credential = TuCredential::parse(contents) .map_err(|error| LifecycleUserError::InvalidCredential(error.to_string()))?; - let (username, public_key) = inspect_credential_account(connection, &credential).await?; - let profile = UserProfile::new( + let (username, public_key, created_at) = + inspect_credential_account(connection, &credential).await?; + let profile = UserProfile::new_with_created_at( credential.user_id, username, None, public_key, hex_hash(contents), String::new(), + created_at, ); - write_user_credential(&profile.username, &credential.to_canonical_string()) - .map_err(|error| LifecycleUserError::LocalPersistence(error.to_string()))?; pending_operations::upsert(&PendingUserOperation { user_id: profile.user_id, operation: PendingUserOperationKind::Attach, @@ -191,6 +197,13 @@ pub async fn attach_user_from_tu( created_at: now_millis(), }) .map_err(|error| LifecycleUserError::LocalPersistence(error.to_string()))?; + write_user_credential(&profile.username, &credential.to_canonical_string()) + .map_err(|error| LifecycleUserError::LocalPersistence(error.to_string()))?; + pending_operations::update_phase( + profile.user_id, + PendingUserOperationPhase::CredentialWritten, + ) + .map_err(|error| LifecycleUserError::LocalPersistence(error.to_string()))?; if let Err(error) = credential_proof( connection, &credential, From 62bd44d7d0ee116bf447517b4fa7d15a8915b01b Mon Sep 17 00:00:00 2001 From: Rasensprenger Date: Sun, 30 Aug 2026 23:01:04 +0300 Subject: [PATCH 3/3] Update Rust crate rand to 0.10 --- Cargo.lock | 43 +---------------------------------------- communities/Cargo.toml | 2 +- iota-storage/Cargo.toml | 2 +- 3 files changed, 3 insertions(+), 44 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e506091..92d3dbe 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2247,7 +2247,7 @@ dependencies = [ "json", "once_cell", "r2d2", - "rand 0.8.7", + "rand 0.10.2", "rusqlite", "serde", "serde_yaml", @@ -3416,15 +3416,6 @@ version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "efca4c95a19a79d1c98f791f10aebd5c1363b473244630bb7dbde1dc98455a24" -[[package]] -name = "ppv-lite86" -version = "0.2.21" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" -dependencies = [ - "zerocopy", -] - [[package]] name = "proc-macro2" version = "1.0.107" @@ -3532,8 +3523,6 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "22f6172bdec972074665ed81ed53b71da00bfc44b65a753cfde883ec4c702a1a" dependencies = [ - "libc", - "rand_chacha", "rand_core 0.6.4", ] @@ -3548,16 +3537,6 @@ dependencies = [ "rand_core 0.10.1", ] -[[package]] -name = "rand_chacha" -version = "0.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" -dependencies = [ - "ppv-lite86", - "rand_core 0.6.4", -] - [[package]] name = "rand_core" version = "0.6.4" @@ -5457,26 +5436,6 @@ dependencies = [ "synstructure", ] -[[package]] -name = "zerocopy" -version = "0.8.56" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "556764e583adb45a9f8d413c2a147fa7e8d821e48e12b14fd560b607998b75eb" -dependencies = [ - "zerocopy-derive", -] - -[[package]] -name = "zerocopy-derive" -version = "0.8.56" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f2ab42fc20575779bd240faa45f94a74256f755c0fa9e89f0ede20d91d0cdfc1" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.119", -] - [[package]] name = "zerofrom" version = "0.1.8" diff --git a/communities/Cargo.toml b/communities/Cargo.toml index 36ef42e..0b3ad25 100644 --- a/communities/Cargo.toml +++ b/communities/Cargo.toml @@ -8,7 +8,7 @@ mtp = { git = "https://git.methanium.net/Methanium/mtp.git" } iota-util = { path = "../iota-util" } futures = "*" -rand = "0.8" +rand = "0.10" rand_core = { version = "0.6", features = ["getrandom", "std"] } sha2 = "0.11.0" tokio = { version = "1.50.0", features = ["full"] } diff --git a/iota-storage/Cargo.toml b/iota-storage/Cargo.toml index 9f91c52..5541ef4 100644 --- a/iota-storage/Cargo.toml +++ b/iota-storage/Cargo.toml @@ -15,6 +15,6 @@ r2d2 = "0.8" serde = { version = "1", features = ["derive"] } serde_yaml = "0.9" thiserror = "2" -rand = "0.8" +rand = "0.10" rusqlite = "0.40.0" tokio = { version = "1.50.0", features = ["full"] }