diff --git a/Cargo.lock b/Cargo.lock index 34f2c00..bcdc7ad 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -174,7 +174,7 @@ dependencies = [ "foldhash", "futures-core", "futures-util", - "impl-more 0.3.5", + "impl-more 0.3.7", "itoa", "language-tags", "log", @@ -866,6 +866,12 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "core_detect" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f8f80099a98041a3d1622845c271458a2d73e688351bf3cb999266764b81d48" + [[package]] name = "cpufeatures" version = "0.2.17" @@ -1278,11 +1284,17 @@ checksum = "252afb9ae5eaa683babdc6a068b3f5726eb19e05070c731f9b2a23a7c3e8ed34" [[package]] name = "encoding_rs" -version = "0.8.35" +version = "0.8.40" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "75030f3c4f45dafd7586dd6780965a8c7e8e285a5ecb86713e63a79c5b2766f3" +checksum = "2a7a45518d2863d18aa47f4a0cf9faec2aa4304cc09df5e41299f276b3ad135e" dependencies = [ "cfg-if", + "core_detect", + "multiversion", + "multiversion_no_op", + "rustversion", + "scopeguard", + "simdutf8", ] [[package]] @@ -1696,9 +1708,9 @@ dependencies = [ [[package]] name = "hashlink" -version = "0.12.1" +version = "0.12.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "32069d97bb81e38fa67eab65e3393bf804bb85969f2bc06bf13f64aef5aba248" +checksum = "a596f1b20ed2cc5ecac41a164aaebc7258057060f06c0cf7a2ba3991ee7990fb" dependencies = [ "hashbrown 0.17.1", ] @@ -2018,9 +2030,9 @@ checksum = "e8a5a9a0ff0086c7a148acb942baaabeadf9504d10400b5a05645853729b9cd2" [[package]] name = "impl-more" -version = "0.3.5" +version = "0.3.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "277ff51754a3f68f12f58446c5d006aa8baa4914ea273cce24a599cfaff33d4f" +checksum = "edaff2ce006342d4d0e00fae676f7082dada44a203560629ba18ea50a19277bb" [[package]] name = "indexmap" @@ -2896,6 +2908,34 @@ dependencies = [ "tracing", ] +[[package]] +name = "multiversion" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7edb7f0ff51249dfda9ab96b5823695e15a052dc15074c9dbf3d118afaf2c201" +dependencies = [ + "multiversion-macros", + "target-features", +] + +[[package]] +name = "multiversion-macros" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b093064383341eb3271f42e381cb8f10a01459478446953953c75d24bd339fc0" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", + "target-features", +] + +[[package]] +name = "multiversion_no_op" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "743fb55ba31b18fb1ecef6bdc9aa2743314978ac084044301a7eee33fb99a20d" + [[package]] name = "nix" version = "0.29.0" @@ -3009,9 +3049,9 @@ dependencies = [ [[package]] name = "octets" -version = "0.3.6" +version = "0.3.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "866cb5af6f3aa3c1b44c3c2d79d22165fbb1b102e1b3fb499864bfe34736ec4b" +checksum = "4d59d30d3ad7f7e1c9a66e51c9cb71ae1c794b28ffe7053fd785defb0d180069" [[package]] name = "oid-registry" @@ -3899,9 +3939,9 @@ dependencies = [ [[package]] name = "rustls" -version = "0.23.43" +version = "0.23.44" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" +checksum = "6725596c3f2c3a0aef021139e145d4eafe314a6623e4680ca83852b2c67ab2ba" dependencies = [ "aws-lc-rs", "log", @@ -4467,6 +4507,12 @@ dependencies = [ "libc", ] +[[package]] +name = "target-features" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c1bbb9f3c5c463a01705937a24fdabc5047929ac764b2d5b9cf681c1f5041ed5" + [[package]] name = "tempfile" version = "3.27.0" diff --git a/client/src/client_connection.rs b/client/src/client_connection.rs index d9fd8c4..2e30bb7 100644 --- a/client/src/client_connection.rs +++ b/client/src/client_connection.rs @@ -210,15 +210,21 @@ impl ClientConnection { return; } - if cv.is_type(CommunicationType::ClientConnected) { - self.send_message(&message_handlers::handle_client_connected(&cv)) - .await; + if cv.is_type(CommunicationType::ClientStateGet) { + // Durable client synchronization is routed through Omikron, which + // is the authority that assigns the numeric transport SessionId. + self.send_message( + &CommunicationValue::new(CommunicationType::ErrorInvalidData).with_request_id(&cv), + ) + .await; return; } if cv.is_type(CommunicationType::ClientStateAck) { - self.send_message(&message_handlers::handle_client_state_ack(&cv)) - .await; + self.send_message( + &CommunicationValue::new(CommunicationType::ErrorInvalidData).with_request_id(&cv), + ) + .await; return; } diff --git a/iota-connection/src/message_handlers.rs b/iota-connection/src/message_handlers.rs index 47af01d..e331f78 100644 --- a/iota-connection/src/message_handlers.rs +++ b/iota-connection/src/message_handlers.rs @@ -17,6 +17,17 @@ use std::sync::atomic::{AtomicU32, Ordering}; static NEXT_NOTIFICATION_ID: AtomicU32 = AtomicU32::new(1); +fn valid_device_id(id: &str) -> bool { + id.len() == 36 + && id + .chars() + .enumerate() + .all(|(index, character)| match index { + 8 | 13 | 18 | 23 => character == '-', + _ => character.is_ascii_hexdigit(), + }) +} + fn next_notification_id() -> u32 { NEXT_NOTIFICATION_ID.fetch_add(1, Ordering::Relaxed).max(1) } @@ -952,7 +963,7 @@ fn sync_error(cv: &CommunicationValue) -> CommunicationValue { } /// The sender is authenticated by MTP; a UserId embedded by a client is never trusted here. -pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue { +pub fn handle_client_state_get(cv: &CommunicationValue) -> CommunicationValue { use iota_storage::util::sync::{self, CACHE_SCHEMA_VERSION}; let user_id = match required_sender_id(cv) { Ok(id) if id > 0 => id, @@ -962,21 +973,31 @@ pub fn handle_client_connected(cv: &CommunicationValue) -> CommunicationValue { Some(id) if id > 0 => id, _ => return sync_error(cv), }; + let device_id = match cv.get_data(DataType::DeviceId).and_then(DataValue::as_str) { + Some(id) if valid_device_id(id) => id.to_owned(), + _ => return sync_error(cv), + }; let reported_version = match data_i64(cv, DataType::VersionNumber) { Some(version) if version >= 0 => version, _ => return sync_error(cv), }; - let cache_valid = cv.get_data(DataType::CacheValid).as_bool().unwrap_or(false); - let schema = data_i64(cv, DataType::CacheSchemaVersion).unwrap_or(0); + let cache_valid = match cv.get_data(DataType::CacheValid).as_bool() { + Some(value) => value, + None => return sync_error(cv), + }; + let schema = match data_i64(cv, DataType::CacheSchemaVersion) { + Some(value) if value >= 0 => value, + _ => return sync_error(cv), + }; let head = match sync::head(user_id) { Ok(version) => version, Err(_) => return sync_error(cv), }; - let known_session = sync::has_session(user_id, session_id).unwrap_or(false); - let acknowledged_version = sync::acknowledged_version(user_id, session_id).unwrap_or(None); + let known_device = sync::has_device(user_id, &device_id).unwrap_or(false); + let acknowledged_version = sync::acknowledged_version(user_id, &device_id).unwrap_or(None); let full = !cache_valid || reported_version == 0 - || !known_session + || !known_device || acknowledged_version.is_some_and(|version| reported_version < version) || reported_version > head || schema != CACHE_SCHEMA_VERSION; @@ -1181,12 +1202,17 @@ pub fn handle_client_state_ack(cv: &CommunicationValue) -> CommunicationValue { Some(id) if id > 0 => id, _ => return sync_error(cv), }; + let device_id = match cv.get_data(DataType::DeviceId).and_then(DataValue::as_str) { + Some(id) if valid_device_id(id) => id.to_owned(), + _ => return sync_error(cv), + }; let version = match data_i64(cv, DataType::VersionNumber) { Some(version) if version >= 0 => version, _ => return sync_error(cv), }; if iota_storage::util::client_message_delivery::acknowledge_client_state( user_id, + &device_id, session_id, version, CACHE_SCHEMA_VERSION, @@ -1874,7 +1900,24 @@ fn blob_mutation_response( ) } -fn blob_changed(user_id: i64, blob_id: String, revision: i64, deleted: bool) -> CommunicationValue { +fn blob_changed(blob: &UserBlob) -> CommunicationValue { + CommunicationValue::new(CommunicationType::UserBlobChanged) + .with_id(next_notification_id()) + .with_receiver(sender_wire_id(blob.user_id)) + .add_typed_default(DataType::BlobId, DataValue::Str(blob.blob_id.clone())) + .add_typed_default(DataType::Blob, DataValue::Bytes(blob.blob.clone())) + .add_typed_default( + DataType::VersionNumber, + DataValue::SignedNumber(blob.revision.into()), + ) + .add_typed_default( + DataType::UpdatedAt, + DataValue::SignedNumber(blob.updated_at.into()), + ) + .add_typed_default(DataType::Deleted, DataValue::Bool(false)) +} + +fn blob_deleted(user_id: i64, blob_id: String, revision: i64) -> CommunicationValue { CommunicationValue::new(CommunicationType::UserBlobChanged) .with_id(next_notification_id()) .with_receiver(sender_wire_id(user_id)) @@ -1883,7 +1926,7 @@ fn blob_changed(user_id: i64, blob_id: String, revision: i64, deleted: bool) -> DataType::VersionNumber, DataValue::SignedNumber(revision.into()), ) - .add_typed_default(DataType::Deleted, DataValue::Bool(deleted)) + .add_typed_default(DataType::Deleted, DataValue::Bool(true)) } fn blob_request(cv: &CommunicationValue) -> Result<(i64, String), CommunicationValue> { @@ -2136,12 +2179,7 @@ pub fn handle_user_blob_put(cv: &CommunicationValue) -> BlobMutation { match user_blobs::put(user_id, &blob_id, &blob, expected_revision) { Ok(stored) => BlobMutation { response: blob_mutation_response(cv, CommunicationType::UserBlobPut, &stored), - changed: Some(blob_changed( - user_id, - stored.blob_id.clone(), - stored.revision, - false, - )), + changed: Some(blob_changed(&stored)), }, Err(iota_storage::storage_error::StorageError::RevisionConflict) => BlobMutation { response: error_response(cv, CommunicationType::ErrorInvalidData), @@ -2184,7 +2222,7 @@ pub fn handle_user_blob_delete(cv: &CommunicationValue) -> BlobMutation { ), changed: deleted .changed - .then(|| blob_changed(user_id, blob_id, deleted.revision, true)), + .then(|| blob_deleted(user_id, blob_id, deleted.revision)), }, Ok(None) => BlobMutation { response: CommunicationValue::new(CommunicationType::UserBlobDelete) diff --git a/iota-storage/src/users/user_manager.rs b/iota-storage/src/users/user_manager.rs index 173f77f..5fedca7 100644 --- a/iota-storage/src/users/user_manager.rs +++ b/iota-storage/src/users/user_manager.rs @@ -450,6 +450,10 @@ fn purge_database_rows(user_id: i64) -> Result<(), crate::storage_error::Storage "DELETE FROM client_sync_state WHERE user_id = ?1", params![user_id], )?; + tx.execute( + "DELETE FROM client_device_sync_state WHERE user_id = ?1", + params![user_id], + )?; tx.execute( "DELETE FROM client_message_deliveries WHERE user_id = ?1", params![user_id], diff --git a/iota-storage/src/util/client_message_delivery.rs b/iota-storage/src/util/client_message_delivery.rs index baa4f5f..e0f3975 100644 --- a/iota-storage/src/util/client_message_delivery.rs +++ b/iota-storage/src/util/client_message_delivery.rs @@ -41,11 +41,12 @@ pub fn acknowledge_sync_delivery( /* Acknowledge a state-sync response and delete only its recorded visible messages atomically. */ pub fn acknowledge_client_state( user_id: i64, + device_id: &str, session_id: i64, version: i64, cache_schema_version: i64, ) -> Result<(), StorageError> { - if user_id <= 0 || session_id <= 0 || version < 0 { + if user_id <= 0 || device_id.is_empty() || session_id <= 0 || version < 0 { return Err(StorageError::Other("invalid sync acknowledgement".into())); } db::with_immediate_transaction(|tx| { @@ -70,7 +71,7 @@ pub fn acknowledge_client_state( })? .collect::, _>>()? }; - tx.execute("INSERT INTO client_sync_state (user_id, session_id, acknowledged_version, cache_schema_version, updated_at) VALUES (?1, ?2, ?3, ?4, ?5) ON CONFLICT(user_id, session_id) DO UPDATE SET acknowledged_version = MAX(acknowledged_version, excluded.acknowledged_version), cache_schema_version = excluded.cache_schema_version, updated_at = excluded.updated_at", params![user_id, session_id, version, cache_schema_version, sync::now_millis()])?; + tx.execute("INSERT INTO client_device_sync_state (user_id, device_id, acknowledged_version, cache_schema_version, updated_at) VALUES (?1, ?2, ?3, ?4, ?5) ON CONFLICT(user_id, device_id) DO UPDATE SET acknowledged_version = MAX(acknowledged_version, excluded.acknowledged_version), cache_schema_version = excluded.cache_schema_version, updated_at = excluded.updated_at", params![user_id, device_id, version, cache_schema_version, sync::now_millis()])?; if message_storage_policy::get_in_tx(tx, user_id)?.history_mode == message_storage_policy::MessageHistoryMode::DeleteAfterClientDelivery { diff --git a/iota-storage/src/util/db.rs b/iota-storage/src/util/db.rs index 025cf84..0b7553a 100644 --- a/iota-storage/src/util/db.rs +++ b/iota-storage/src/util/db.rs @@ -887,6 +887,22 @@ fn run_migrations_on_connection(conn: &Connection) -> Result<(), StorageError> { conn.pragma_update(None, "user_version", 25)?; } + if current_version < 26 { + conn.execute_batch( + r#" + CREATE TABLE IF NOT EXISTS client_device_sync_state ( + user_id INTEGER NOT NULL, + device_id TEXT NOT NULL, + acknowledged_version INTEGER NOT NULL, + cache_schema_version INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + PRIMARY KEY (user_id, device_id) + ); + "#, + )?; + conn.pragma_update(None, "user_version", 26)?; + } + Ok(()) } @@ -956,7 +972,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, 25); + assert_eq!(version, 26); 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")?; @@ -975,11 +991,12 @@ 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, 25); + assert_eq!(version, 26); for table in [ "sync_heads", "sync_events", "client_sync_state", + "client_device_sync_state", "user_residency", "relay_replay", "pending_relays", @@ -1021,7 +1038,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, 25); + assert_eq!(version, 26); for column in [ "id", "user_id", @@ -1116,7 +1133,7 @@ mod tests { )?; let version: i64 = conn.pragma_query_value(None, "user_version", |row| row.get(0))?; assert_eq!(preserved, "remote_committed"); - assert_eq!(version, 25); + assert_eq!(version, 26); Ok(()) } } diff --git a/iota-storage/src/util/sync.rs b/iota-storage/src/util/sync.rs index e4aa539..9d31474 100644 --- a/iota-storage/src/util/sync.rs +++ b/iota-storage/src/util/sync.rs @@ -114,23 +114,23 @@ pub fn head(user_id: i64) -> Result { }) } -pub fn has_session(user_id: i64, session_id: i64) -> Result { +pub fn has_device(user_id: i64, device_id: &str) -> Result { db::with_db(|conn| { Ok(conn .query_row( - "SELECT 1 FROM client_sync_state WHERE user_id = ?1 AND session_id = ?2", - params![user_id, session_id], + "SELECT 1 FROM client_device_sync_state WHERE user_id = ?1 AND device_id = ?2", + params![user_id, device_id], |_| Ok(()), ) .is_ok()) }) } -pub fn acknowledged_version(user_id: i64, session_id: i64) -> Result, StorageError> { +pub fn acknowledged_version(user_id: i64, device_id: &str) -> Result, StorageError> { db::with_db(|conn| { conn.query_row( - "SELECT acknowledged_version FROM client_sync_state WHERE user_id = ?1 AND session_id = ?2", - params![user_id, session_id], + "SELECT acknowledged_version FROM client_device_sync_state WHERE user_id = ?1 AND device_id = ?2", + params![user_id, device_id], |row| row.get(0), ) .optional() @@ -138,33 +138,6 @@ pub fn acknowledged_version(user_id: i64, session_id: i64) -> Result }) } -pub fn acknowledge( - user_id: i64, - session_id: i64, - version: i64, - cache_schema_version: i64, -) -> Result<(), StorageError> { - if user_id <= 0 || session_id <= 0 || version < 0 { - return Err(StorageError::Other("invalid sync acknowledgement".into())); - } - db::with_db(|conn| { - let head = conn - .query_row( - "SELECT version FROM sync_heads WHERE user_id = ?1", - [user_id], - |r| r.get(0), - ) - .unwrap_or(0); - if version > head { - return Err(StorageError::Other( - "acknowledgement is ahead of head".into(), - )); - } - conn.execute("INSERT INTO client_sync_state (user_id, session_id, acknowledged_version, cache_schema_version, updated_at) VALUES (?1, ?2, ?3, ?4, ?5) ON CONFLICT(user_id, session_id) DO UPDATE SET acknowledged_version = MAX(acknowledged_version, excluded.acknowledged_version), cache_schema_version = excluded.cache_schema_version, updated_at = excluded.updated_at", params![user_id, session_id, version, cache_schema_version, now_millis()])?; - Ok(()) - }) -} - /* Retain deltas until every non-stale client has acknowledged them. */ pub fn compact_user_sync_state(user_id: i64) -> Result { if user_id <= 0 { @@ -173,15 +146,15 @@ pub fn compact_user_sync_state(user_id: i64) -> Result = tx.query_row( - "SELECT MIN(acknowledged_version) FROM client_sync_state WHERE user_id = ?1", + "SELECT MIN(acknowledged_version) FROM client_device_sync_state WHERE user_id = ?1", [user_id], |row| row.get(0), )?; diff --git a/mtp-type-maps b/mtp-type-maps index 154096c..2de3be6 160000 --- a/mtp-type-maps +++ b/mtp-type-maps @@ -1 +1 @@ -Subproject commit 154096cd198d6ccd0663e5e701643ebc859fafab +Subproject commit 2de3be6016410304d6505d12ef37c35e63a9c731 diff --git a/omikron-connector/src/omikron_connection.rs b/omikron-connector/src/omikron_connection.rs index 3258883..d0531a7 100644 --- a/omikron-connector/src/omikron_connection.rs +++ b/omikron-connector/src/omikron_connection.rs @@ -2321,7 +2321,7 @@ impl OmikronConnection { dispatch!(LoadAppData, handle_load_app_data); dispatch!(CreateApp, handle_create_app); dispatch!(DeleteApp, handle_delete_app); - dispatch!(ClientConnected, handle_client_connected); + dispatch!(ClientStateGet, handle_client_state_get); dispatch!(ClientStateAck, handle_client_state_ack); dispatch!(ReadNotification, handle_read_notification); dispatch!(MessageEdit, handle_message_edit); @@ -2622,8 +2622,8 @@ impl OmikronConnection { .await; } - async fn handle_client_connected(self: Arc, cv: &CommunicationValue) { - let response = message_handlers::handle_client_connected(cv); + async fn handle_client_state_get(self: Arc, cv: &CommunicationValue) { + let response = message_handlers::handle_client_state_get(cv); let user_id = cv .require_sender() .ok() @@ -2663,7 +2663,7 @@ impl OmikronConnection { } if let Err(error) = self.send_message(&response).await { - log!("ClientConnected response delivery failed: {}", error); + log!("ClientStateGet response delivery failed: {}", error); } }