diff --git a/crates/r3akt-rch-core/migrations/0003_topic_subscription_corrections.sql b/crates/r3akt-rch-core/migrations/0003_topic_subscription_corrections.sql new file mode 100644 index 0000000..1318c64 --- /dev/null +++ b/crates/r3akt-rch-core/migrations/0003_topic_subscription_corrections.sql @@ -0,0 +1,3 @@ +-- Topic subscription corrections: support topic-first fan-out reads. +CREATE INDEX IF NOT EXISTS idx_rch_subscribers_topic_node + ON rch_subscribers (topic_id, node_id); diff --git a/crates/r3akt-rch-core/src/lib.rs b/crates/r3akt-rch-core/src/lib.rs index 9eef56d..7148109 100644 --- a/crates/r3akt-rch-core/src/lib.rs +++ b/crates/r3akt-rch-core/src/lib.rs @@ -393,7 +393,9 @@ const RCH_ROLE_BUNDLES: &[RchRoleBundleDefinition] = &[ const RCH_SQLITE_MIGRATION_SQL: &str = include_str!("../migrations/0001_rch_core_snapshot.sql"); const RCH_SQLITE_MIGRATION_2_SQL: &str = include_str!("../migrations/0002_ordered_migrations.sql"); -const RCH_SQLITE_SCHEMA_VERSION: &str = "2"; +const RCH_SQLITE_MIGRATION_3_SQL: &str = + include_str!("../migrations/0003_topic_subscription_corrections.sql"); +const RCH_SQLITE_SCHEMA_VERSION: &str = "3"; const RCH_SQLITE_READ_BUSY_TIMEOUT_MS: u64 = 250; const RCH_SQLITE_WRITE_BUSY_TIMEOUT_MS: u64 = 1_000; const RCH_SQLITE_ADMIN_BUSY_TIMEOUT_MS: u64 = 30_000; @@ -1070,7 +1072,6 @@ pub struct RchCommandOutcome { #[derive(Debug, Default)] pub struct RchCore { topics: HashMap, - subscriptions: HashSet<(String, String)>, subscribers: HashMap<(String, String), SubscriberRecord>, messages: Vec, clients: HashMap, @@ -2078,10 +2079,16 @@ impl RchSqliteStore { } pub fn delete_topic(&mut self, topic_id: &str) -> Result<(), RchCoreError> { - self.connection.execute( + let transaction = self.connection.transaction()?; + transaction.execute( "DELETE FROM rch_topics WHERE topic_id = ?1", params![topic_id], )?; + transaction.execute( + "DELETE FROM rch_subscribers WHERE topic_id = ?1", + params![topic_id], + )?; + transaction.commit()?; Ok(()) } @@ -2767,12 +2774,18 @@ impl RchSqliteStore { [], |row| row.get::<_, bool>(0), )?; - if migration_2_applied { + let migration_3_applied = has_migration_history + && self.connection.query_row( + "SELECT EXISTS(SELECT 1 FROM rch_schema_migrations WHERE version = 3)", + [], + |row| row.get::<_, bool>(0), + )?; + if migration_3_applied { return Ok(()); } if has_existing_schema { self.integrity_check()?; - self.backup_before_migration(2)?; + self.backup_before_migration(if migration_2_applied { 3 } else { 2 })?; } self.connection.execute_batch("BEGIN IMMEDIATE;")?; let migration_result = (|| -> Result<(), RchCoreError> { @@ -2789,60 +2802,67 @@ impl RchSqliteStore { VALUES (1, 'rch_core_snapshot', ?1)", [utc_now_ms()], )?; - if !sqlite_table_has_column( - &self.connection, - "rch_checklist_feed_publications", - "published_ts_ms", - )? { - self.connection.execute( - "ALTER TABLE rch_checklist_feed_publications + if !migration_2_applied { + if !sqlite_table_has_column( + &self.connection, + "rch_checklist_feed_publications", + "published_ts_ms", + )? { + self.connection.execute( + "ALTER TABLE rch_checklist_feed_publications ADD COLUMN published_ts_ms INTEGER NOT NULL DEFAULT 0", - [], - )?; - } - if !sqlite_table_has_column(&self.connection, "rch_checklist_columns", "display_order")? - { - self.connection.execute( - "ALTER TABLE rch_checklist_columns - ADD COLUMN display_order INTEGER NOT NULL DEFAULT 0", - [], - )?; - } - for (column, definition) in [ - ("delivery_state", "TEXT NOT NULL DEFAULT 'queued'"), - ("dispatch_status", "TEXT NOT NULL DEFAULT 'queued'"), - ("next_attempt_at_ts_ms", "INTEGER"), - ("attempts", "INTEGER NOT NULL DEFAULT 0"), - ("priority", "INTEGER NOT NULL DEFAULT 0"), - ("batch_id", "TEXT"), - ("created_ts_ms", "INTEGER NOT NULL DEFAULT 0"), - ] { - if !sqlite_table_has_column(&self.connection, "rch_messages", column)? { + [], + )?; + } + if !sqlite_table_has_column( + &self.connection, + "rch_checklist_columns", + "display_order", + )? { self.connection.execute( - &format!("ALTER TABLE rch_messages ADD COLUMN {column} {definition}"), + "ALTER TABLE rch_checklist_columns + ADD COLUMN display_order INTEGER NOT NULL DEFAULT 0", [], )?; } - } - if !sqlite_table_has_column(&self.connection, "rch_mission_changes", "mission_uid")? { - self.connection.execute( + for (column, definition) in [ + ("delivery_state", "TEXT NOT NULL DEFAULT 'queued'"), + ("dispatch_status", "TEXT NOT NULL DEFAULT 'queued'"), + ("next_attempt_at_ts_ms", "INTEGER"), + ("attempts", "INTEGER NOT NULL DEFAULT 0"), + ("priority", "INTEGER NOT NULL DEFAULT 0"), + ("batch_id", "TEXT"), + ("created_ts_ms", "INTEGER NOT NULL DEFAULT 0"), + ] { + if !sqlite_table_has_column(&self.connection, "rch_messages", column)? { + self.connection.execute( + &format!("ALTER TABLE rch_messages ADD COLUMN {column} {definition}"), + [], + )?; + } + } + if !sqlite_table_has_column(&self.connection, "rch_mission_changes", "mission_uid")? + { + self.connection.execute( "ALTER TABLE rch_mission_changes ADD COLUMN mission_uid TEXT NOT NULL DEFAULT ''", [], )?; - } - for (column, definition) in [ - ("mission_uid", "TEXT NOT NULL DEFAULT ''"), - ("task_uid", "TEXT NOT NULL DEFAULT ''"), - ("team_member_rns_identity", "TEXT NOT NULL DEFAULT ''"), - ] { - if !sqlite_table_has_column(&self.connection, "rch_assignments", column)? { - self.connection.execute( - &format!("ALTER TABLE rch_assignments ADD COLUMN {column} {definition}"), - [], - )?; } - } - self.connection.execute_batch( + for (column, definition) in [ + ("mission_uid", "TEXT NOT NULL DEFAULT ''"), + ("task_uid", "TEXT NOT NULL DEFAULT ''"), + ("team_member_rns_identity", "TEXT NOT NULL DEFAULT ''"), + ] { + if !sqlite_table_has_column(&self.connection, "rch_assignments", column)? { + self.connection.execute( + &format!( + "ALTER TABLE rch_assignments ADD COLUMN {column} {definition}" + ), + [], + )?; + } + } + self.connection.execute_batch( "CREATE INDEX IF NOT EXISTS idx_rch_messages_queue_due ON rch_messages (delivery_state, dispatch_status, next_attempt_at_ts_ms, priority, id); CREATE INDEX IF NOT EXISTS idx_rch_messages_batch @@ -2881,13 +2901,22 @@ impl RchSqliteStore { WHERE delivery_state = 'queued'; CREATE INDEX IF NOT EXISTS idx_rch_outbound_jobs_batch ON rch_outbound_jobs (batch_id);", - )?; - self.connection.execute_batch(RCH_SQLITE_MIGRATION_2_SQL)?; - self.connection.execute( - "INSERT INTO rch_schema_migrations (version, name, applied_ts_ms) - VALUES (2, 'ordered_migrations', ?1)", - [utc_now_ms()], - )?; + )?; + self.connection.execute_batch(RCH_SQLITE_MIGRATION_2_SQL)?; + self.connection.execute( + "INSERT INTO rch_schema_migrations (version, name, applied_ts_ms) + VALUES (2, 'ordered_migrations', ?1)", + [utc_now_ms()], + )?; + } + if !migration_3_applied { + self.connection.execute_batch(RCH_SQLITE_MIGRATION_3_SQL)?; + self.connection.execute( + "INSERT INTO rch_schema_migrations (version, name, applied_ts_ms) + VALUES (3, 'topic_subscription_corrections', ?1)", + [utc_now_ms()], + )?; + } self.connection.execute( "INSERT OR REPLACE INTO rch_settings (setting_key, setting_value) VALUES ('schema_version', ?1)", @@ -4080,16 +4109,82 @@ impl RchCore { records } + fn validate_subscriber_invariants(&self) -> Result<(), RchCoreError> { + let mut normalized_pairs = HashSet::new(); + for ((node_id, topic_id), subscriber) in &self.subscribers { + let normalized_node_id = + normalize_subscriber_node_id(Some(node_id)).ok_or_else(|| { + RchCoreError::InvalidPayload( + "subscriber invariant violation: destination is empty".to_string(), + ) + })?; + let normalized_topic_id = normalize_subscriber_topic_id(topic_id).ok_or_else(|| { + RchCoreError::InvalidPayload( + "subscriber invariant violation: topic_id is invalid".to_string(), + ) + })?; + if normalized_node_id != *node_id + || normalized_topic_id != *topic_id + || subscriber.node_id != *node_id + || subscriber.topic_id != *topic_id + { + return Err(RchCoreError::InvalidPayload( + "subscriber invariant violation: key and record are not normalized or do not match" + .to_string(), + )); + } + // The compatibility /Subscriber routes historically persist + // topic-less and orphaned rows. New core subscriptions still + // require an existing topic in `subscribe`, but loading those + // rows must remain lossless and must not make R3AKT reads fail. + if !normalized_pairs.insert((normalized_node_id, normalized_topic_id)) { + return Err(RchCoreError::InvalidPayload( + "subscriber invariant violation: duplicate normalized destination/topic pair" + .to_string(), + )); + } + } + Ok(()) + } + #[allow(clippy::too_many_lines)] pub fn from_snapshot(snapshot: RchCoreSnapshot) -> Result { let mut core = Self::new(); - for topic in snapshot.topics { - core.topics.insert(topic.topic_id.clone(), topic); + for mut topic in snapshot.topics { + let topic_id = normalize_topic_id(Some(&topic.topic_id)).ok_or_else(|| { + RchCoreError::InvalidPayload( + "topic invariant violation: topic_id is empty".to_string(), + ) + })?; + topic.topic_id.clone_from(&topic_id); + if core.topics.insert(topic_id.clone(), topic).is_some() { + return Err(RchCoreError::InvalidPayload(format!( + "topic invariant violation: duplicate normalized topic '{topic_id}'" + ))); + } } - for subscriber in snapshot.subscribers { - let key = (subscriber.node_id.clone(), subscriber.topic_id.clone()); - core.subscriptions.insert(key.clone()); - core.subscribers.insert(key, subscriber); + for mut subscriber in snapshot.subscribers { + let node_id = + normalize_subscriber_node_id(Some(&subscriber.node_id)).ok_or_else(|| { + RchCoreError::InvalidPayload( + "subscriber invariant violation: destination is empty".to_string(), + ) + })?; + let topic_id = + normalize_subscriber_topic_id(&subscriber.topic_id).ok_or_else(|| { + RchCoreError::InvalidPayload( + "subscriber invariant violation: topic_id is invalid".to_string(), + ) + })?; + subscriber.node_id.clone_from(&node_id); + subscriber.topic_id.clone_from(&topic_id); + let key = (node_id, topic_id); + if core.subscribers.insert(key, subscriber).is_some() { + return Err(RchCoreError::InvalidPayload( + "subscriber invariant violation: duplicate normalized destination/topic pair" + .to_string(), + )); + } } core.messages = snapshot.messages; for client in snapshot.clients { @@ -4235,6 +4330,7 @@ impl RchCore { ); } core.authorization_required = snapshot.authorization_required; + core.validate_subscriber_invariants()?; Ok(core) } @@ -8886,6 +8982,8 @@ impl RchCore { attachment.updated_ts_ms = utc_now_ms(); } } + self.subscribers + .retain(|(_, subscribed_topic_id), _| subscribed_topic_id != &topic_id); Ok(topic) } @@ -8928,9 +9026,10 @@ impl RchCore { if !self.topics.contains_key(topic_id) { return Err(RchCoreError::TopicNotFound); } + let subscriber_id = normalize_subscriber_node_id(Some(subscriber_id)) + .ok_or_else(|| RchCoreError::InvalidPayload("Destination is required".to_string()))?; let now = utc_now_ms(); - let key = (subscriber_id.to_string(), topic_id.to_string()); - self.subscriptions.insert(key.clone()); + let key = (subscriber_id.clone(), topic_id.to_string()); self.subscribers .entry(key) .and_modify(|record| { @@ -8939,7 +9038,7 @@ impl RchCore { record.metadata = metadata.clone(); }) .or_insert_with(|| SubscriberRecord { - node_id: subscriber_id.to_string(), + node_id: subscriber_id, topic_id: topic_id.to_string(), first_seen_ts_ms: now, last_seen_ts_ms: now, @@ -8954,6 +9053,8 @@ impl RchCore { args: &Value, ) -> Result { let subscriber_id = required_text(args, &["subscriber_id", "SubscriberID"])?; + let subscriber_id = normalize_subscriber_node_id(Some(&subscriber_id)) + .ok_or_else(|| RchCoreError::InvalidPayload("SubscriberID is required".to_string()))?; let existing_key = self .subscribers .keys() @@ -8962,19 +9063,20 @@ impl RchCore { .ok_or(RchCoreError::TopicNotFound)?; let mut subscriber = self .subscribers - .remove(&existing_key) + .get(&existing_key) + .cloned() .ok_or(RchCoreError::TopicNotFound)?; - self.subscriptions.remove(&existing_key); if let Some(destination) = optional_text(args, &["destination", "Destination"]) { - subscriber.node_id = destination; + subscriber.node_id = + normalize_subscriber_node_id(Some(&destination)).ok_or_else(|| { + RchCoreError::InvalidPayload("Destination is required".to_string()) + })?; } if let Some(topic_id) = optional_text(args, &["topic_id", "TopicID"]) { let topic_id = normalize_topic_id(Some(&topic_id)) .ok_or_else(|| RchCoreError::InvalidPayload("TopicID is required".to_string()))?; if !self.topics.contains_key(&topic_id) { - self.subscribers.insert(existing_key.clone(), subscriber); - self.subscriptions.insert(existing_key); return Err(RchCoreError::TopicNotFound); } subscriber.topic_id = topic_id; @@ -8991,7 +9093,12 @@ impl RchCore { subscriber.last_seen_ts_ms = utc_now_ms(); let updated_key = (subscriber.node_id.clone(), subscriber.topic_id.clone()); - self.subscriptions.insert(updated_key.clone()); + if updated_key != existing_key && self.subscribers.contains_key(&updated_key) { + return Err(RchCoreError::InvalidPayload( + "normalized subscriber destination/topic pair already exists".to_string(), + )); + } + self.subscribers.remove(&existing_key); self.subscribers.insert(updated_key, subscriber.clone()); Ok(subscriber) } @@ -9001,13 +9108,14 @@ impl RchCore { args: &Value, ) -> Result { let subscriber_id = required_text(args, &["subscriber_id", "SubscriberID", "id", "ID"])?; + let subscriber_id = normalize_subscriber_node_id(Some(&subscriber_id)) + .ok_or_else(|| RchCoreError::InvalidPayload("SubscriberID is required".to_string()))?; let key = self .subscribers .keys() .find(|(node_id, _)| node_id == &subscriber_id) .cloned() .ok_or(RchCoreError::TopicNotFound)?; - self.subscriptions.remove(&key); self.subscribers .remove(&key) .ok_or(RchCoreError::TopicNotFound) @@ -10345,6 +10453,27 @@ pub fn normalize_topic_id(value: Option<&str>) -> Option { .or_else(|| Some(text.to_string())) } +fn normalize_subscriber_node_id(value: Option<&str>) -> Option { + let text = value?.trim(); + if text.is_empty() { + return None; + } + if text.len() == 32 && text.bytes().all(|byte| byte.is_ascii_hexdigit()) { + Some(text.to_ascii_lowercase()) + } else { + Some(text.to_string()) + } +} + +fn normalize_subscriber_topic_id(value: &str) -> Option { + let text = value.trim(); + if text.is_empty() { + Some(String::new()) + } else { + normalize_topic_id(Some(text)) + } +} + #[must_use] pub fn normalize_topic_id_bytes(value: &[u8]) -> Option { if value.is_empty() { @@ -13080,7 +13209,7 @@ mod tests { .expect("setting lookup"), None ); - assert_eq!(store.schema_version().expect("schema version"), "2"); + assert_eq!(store.schema_version().expect("schema version"), "3"); } #[test] @@ -13403,6 +13532,8 @@ mod tests { CommandResultStatus::Accepted ); assert_eq!(core.subscribers("mission-1").len(), 1); + core.validate_subscriber_invariants() + .expect("subscriber invariants"); assert_eq!( core.handle_command(&list).result.status, CommandResultStatus::Accepted @@ -13539,9 +13670,9 @@ mod tests { "mission.topic.deleted" ); assert!(core.topics().is_empty()); - let orphaned_subscribers = core.subscribers("mission-1"); - assert_eq!(orphaned_subscribers.len(), 1); - assert_eq!(orphaned_subscribers[0].node_id, "FACEFEED"); + assert!(core.subscribers("mission-1").is_empty()); + core.validate_subscriber_invariants() + .expect("subscriber invariants after topic deletion"); } #[test] @@ -16567,7 +16698,7 @@ mod tests { let mut store = RchSqliteStore::in_memory().expect("sqlite"); core.save_to_sqlite(&mut store).expect("save"); - assert_eq!(store.schema_version().expect("schema version"), "2"); + assert_eq!(store.schema_version().expect("schema version"), "3"); assert_sqlite_snapshot_counts(&store); let mut restored = RchCore::load_from_sqlite(&store) @@ -16611,7 +16742,19 @@ mod tests { .compact_if_free_percent_exceeds(100.0, 100) .expect("no-op compaction"); - assert_eq!(migration_count, 2); + assert_eq!(migration_count, 3); + + let index_exists: bool = store + .connection + .query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'index' AND name = 'idx_rch_subscribers_topic_node')", + [], + |row| row.get(0), + ) + .expect("subscriber topic index"); + assert!(index_exists); + + drop(store); assert!(stats.page_count > 0); assert!(!report.compacted); assert!(!report.requires_incremental_mode_conversion); @@ -16635,48 +16778,6 @@ mod tests { assert_eq!(version, "1"); } - #[test] - fn sqlite_migration_is_additive_for_existing_database_like_python_startup() { - let db_path = std::env::temp_dir().join(format!( - "r3akt-rch-core-additive-migration-{}.db", - Uuid::new_v4() - )); - { - let connection = Connection::open(&db_path).expect("sqlite"); - connection - .execute_batch( - "CREATE TABLE legacy_probe (id TEXT PRIMARY KEY); - INSERT INTO legacy_probe (id) VALUES ('probe-1');", - ) - .expect("legacy table"); - } - - let store = RchSqliteStore::open(&db_path).expect("migrated store"); - assert_eq!(store.schema_version().expect("schema version"), "2"); - drop(store); - - let connection = Connection::open(&db_path).expect("sqlite"); - let legacy_id: String = connection - .query_row( - "SELECT id FROM legacy_probe WHERE id = 'probe-1'", - [], - |row| row.get(0), - ) - .expect("legacy row"); - let rch_topic_table: String = connection - .query_row( - "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'rch_topics'", - [], - |row| row.get(0), - ) - .expect("rch table"); - - assert_eq!(legacy_id, "probe-1"); - assert_eq!(rch_topic_table, "rch_topics"); - drop(connection); - let _ = std::fs::remove_file(db_path); - } - #[test] fn sqlite_migration_adds_feed_publication_timestamp_to_legacy_table() { let db_path = std::env::temp_dir().join(format!( @@ -16701,7 +16802,7 @@ mod tests { } let store = RchSqliteStore::open(&db_path).expect("migrated store"); - assert_eq!(store.schema_version().expect("schema version"), "2"); + assert_eq!(store.schema_version().expect("schema version"), "3"); drop(store); let connection = Connection::open(&db_path).expect("sqlite"); @@ -16927,4 +17028,6 @@ mod tests { assert_eq!(restored.task_skill_requirements().len(), 1); assert_eq!(restored.assignments().len(), 1); } + + include!("topic_subscription_corrections_tests.rs"); } diff --git a/crates/r3akt-rch-core/src/topic_subscription_corrections_tests.rs b/crates/r3akt-rch-core/src/topic_subscription_corrections_tests.rs new file mode 100644 index 0000000..2d5c2b5 --- /dev/null +++ b/crates/r3akt-rch-core/src/topic_subscription_corrections_tests.rs @@ -0,0 +1,221 @@ +#[test] +fn snapshot_rejects_duplicate_normalized_subscriptions_and_preserves_compatibility_rows() { + let mut core = RchCore::new(); + core.handle_command(&command( + "topic.create", + json!({ "topic_id": "mission-1", "topic_path": "mission-1", "topic_name": "Mission 1" }), + )); + core.handle_command(&command( + "topic.subscribe", + json!({ "topic_id": "mission-1", "destination": "FACEFEED" }), + )); + + let mut duplicate = core.snapshot(); + duplicate.subscribers.push(SubscriberRecord { + node_id: " FACEFEED ".to_string(), + topic_id: " mission-1 ".to_string(), + first_seen_ts_ms: 0, + last_seen_ts_ms: 0, + reject_tests: None, + metadata: json!({}), + }); + let duplicate_error = RchCore::from_snapshot(duplicate).expect_err("duplicate rejected"); + assert!( + duplicate_error + .to_string() + .contains("duplicate normalized destination/topic pair") + ); + + let mut orphan = core.snapshot(); + orphan.subscribers[0].topic_id = "missing-topic".to_string(); + let restored_orphan = RchCore::from_snapshot(orphan).expect("orphan preserved"); + assert_eq!( + restored_orphan.snapshot().subscribers[0].topic_id, + "missing-topic" + ); + + let mut topicless = core.snapshot(); + topicless.subscribers[0].topic_id = " ".to_string(); + let restored_topicless = RchCore::from_snapshot(topicless).expect("topicless preserved"); + assert_eq!(restored_topicless.snapshot().subscribers[0].topic_id, ""); +} + +#[test] +fn sqlite_read_snapshot_keeps_topicless_compatibility_subscribers_loadable() { + let db_path = std::env::temp_dir().join(format!( + "r3akt-rch-core-topicless-subscriber-{}.db", + Uuid::new_v4() + )); + let mut store = RchSqliteStore::open(&db_path).expect("sqlite"); + store + .upsert_topic(&TopicRecord { + topic_id: "ops".to_string(), + topic_name: "Ops".to_string(), + topic_path: "ops".to_string(), + topic_description: String::new(), + retention: RetentionPolicy::Persistent, + visibility: Visibility::Public, + created_ts_ms: 1, + last_activity_ts_ms: 1, + }) + .expect("topic"); + store + .upsert_subscriber(&SubscriberRecord { + node_id: "DEST-COMPAT".to_string(), + topic_id: String::new(), + first_seen_ts_ms: 1, + last_seen_ts_ms: 1, + reject_tests: None, + metadata: json!({ "source": "compatibility" }), + }) + .expect("subscriber"); + + let snapshot = store + .load_r3akt_read_snapshot() + .expect("read snapshot"); + let core = RchCore::from_snapshot(snapshot).expect("compatibility snapshot"); + assert_eq!(core.snapshot().subscribers[0].node_id, "DEST-COMPAT"); + assert_eq!(core.snapshot().subscribers[0].topic_id, ""); + + drop(store); + let _ = std::fs::remove_file(db_path); +} +#[test] +fn sqlite_migration_is_additive_for_existing_database_like_python_startup() { + let db_path = std::env::temp_dir().join(format!( + "r3akt-rch-core-additive-migration-{}.db", + Uuid::new_v4() + )); + { + let connection = Connection::open(&db_path).expect("sqlite"); + connection + .execute_batch( + "CREATE TABLE legacy_probe (id TEXT PRIMARY KEY); + INSERT INTO legacy_probe (id) VALUES ('probe-1');", + ) + .expect("legacy table"); + } + + let store = RchSqliteStore::open(&db_path).expect("migrated store"); + assert_eq!(store.schema_version().expect("schema version"), "3"); + let migration_count: i64 = store + .connection + .query_row("SELECT COUNT(*) FROM rch_schema_migrations", [], |row| { + row.get(0) + }) + .expect("migration count"); + assert_eq!(migration_count, 3); + let index_count: i64 = store + .connection + .query_row( + "SELECT COUNT(*) FROM sqlite_master WHERE type = 'index' AND name = 'idx_rch_subscribers_topic_node'", + [], + |row| row.get(0), + ) + .expect("subscriber topic index"); + assert_eq!(index_count, 1); + drop(store); + + let reopened = RchSqliteStore::open(&db_path).expect("reopened store"); + assert_eq!(reopened.schema_version().expect("schema version"), "3"); + let reopened_migration_count: i64 = reopened + .connection + .query_row("SELECT COUNT(*) FROM rch_schema_migrations", [], |row| { + row.get(0) + }) + .expect("reopened migration count"); + assert_eq!(reopened_migration_count, 3); + drop(reopened); + + let connection = Connection::open(&db_path).expect("sqlite"); + let legacy_id: String = connection + .query_row( + "SELECT id FROM legacy_probe WHERE id = 'probe-1'", + [], + |row| row.get(0), + ) + .expect("legacy row"); + let rch_topic_table: String = connection + .query_row( + "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'rch_topics'", + [], + |row| row.get(0), + ) + .expect("rch table"); + + assert_eq!(legacy_id, "probe-1"); + assert_eq!(rch_topic_table, "rch_topics"); + drop(connection); + let _ = std::fs::remove_file(db_path); +} + +#[test] +fn sqlite_migration_preserves_existing_subscriber_payloads() { + let db_path = std::env::temp_dir().join(format!( + "r3akt-rch-core-subscriber-migration-{}.db", + Uuid::new_v4() + )); + let topic = TopicRecord { + topic_id: "mission-legacy".to_string(), + topic_name: "Legacy Mission".to_string(), + topic_path: "mission-legacy".to_string(), + topic_description: String::new(), + retention: RetentionPolicy::Persistent, + visibility: Visibility::Public, + created_ts_ms: 1, + last_activity_ts_ms: 2, + }; + let subscriber = SubscriberRecord { + node_id: "DEST-LEGACY".to_string(), + topic_id: "mission-legacy".to_string(), + first_seen_ts_ms: 3, + last_seen_ts_ms: 4, + reject_tests: Some(1), + metadata: json!({ "source": "legacy" }), + }; + { + let connection = Connection::open(&db_path).expect("sqlite"); + connection + .execute_batch(include_str!("../migrations/0001_rch_core_snapshot.sql")) + .expect("v1 migration"); + connection + .execute_batch(include_str!("../migrations/0002_ordered_migrations.sql")) + .expect("v2 migration"); + connection + .execute( + "INSERT INTO rch_schema_migrations (version, name, applied_ts_ms) VALUES (1, 'rch_core_snapshot', 1), (2, 'ordered_migrations', 2)", + [], + ) + .expect("migration history"); + connection + .execute( + "INSERT INTO rch_topics (topic_id, payload) VALUES (?1, ?2)", + params![ + &topic.topic_id, + encode_msgpack(&topic).expect("topic payload") + ], + ) + .expect("topic payload"); + connection + .execute( + "INSERT INTO rch_subscribers (node_id, topic_id, payload) VALUES (?1, ?2, ?3)", + params![ + &subscriber.node_id, + &subscriber.topic_id, + encode_msgpack(&subscriber).expect("subscriber payload") + ], + ) + .expect("subscriber payload"); + } + + let store = RchSqliteStore::open(&db_path).expect("migrated store"); + assert_eq!(store.schema_version().expect("schema version"), "3"); + let snapshot = store + .load_snapshot() + .expect("load snapshot") + .expect("snapshot"); + assert_eq!(snapshot.topics, vec![topic]); + assert_eq!(snapshot.subscribers, vec![subscriber]); + drop(store); + let _ = std::fs::remove_file(db_path); +} diff --git a/crates/r3akt-rch-server/src/lib.rs b/crates/r3akt-rch-server/src/lib.rs index b0a63c5..9dd428e 100644 --- a/crates/r3akt-rch-server/src/lib.rs +++ b/crates/r3akt-rch-server/src/lib.rs @@ -40,10 +40,12 @@ mod auth; mod rem_team_routing; mod reticulumd_inbound; +mod topic_diagnostics; use rem_team_routing::{ fanout_mission_sync_response_to_team, send_mission_sync_response_to_source, }; +use topic_diagnostics::{apply_config_text, topic_subscription_diagnostics}; use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; use std::future::pending; @@ -248,6 +250,7 @@ pub struct AppState { lxmf_zmq_response_endpoint: Option>, lxmf_zmq_data_plane: Option>, rch_service_identity_config: Option>, + identity_announce_update_error: Arc>>, rch_source_assertion: Option>, runtime_control: Arc>, managed_reticulumd: Arc>, @@ -523,6 +526,7 @@ impl Default for AppState { lxmf_zmq_response_endpoint: None, lxmf_zmq_data_plane: None, rch_service_identity_config: None, + identity_announce_update_error: Arc::default(), rch_source_assertion: None, runtime_control: Arc::default(), managed_reticulumd: Arc::default(), @@ -10305,6 +10309,7 @@ fn runtime_diagnostics_payload(state: &AppState) -> Result { .clone(); let outbound_delivery = outbound_delivery_diagnostics_snapshot(state)?; let runtime_metrics = runtime_metrics_compat_payload(state, &outbound_delivery); + let topic_diagnostics = topic_subscription_diagnostics(state)?; Ok(json!({ "runtime": "rust", "status": control.status, @@ -10316,6 +10321,14 @@ fn runtime_diagnostics_payload(state: &AppState) -> Result { "shutdown_requested": control.shutdown_requested, "reticulumd_rpc_configured": state.reticulumd_rpc_endpoint.is_some(), "reticulumd_source_configured": state.reticulumd_source.is_some(), + "topic_count": topic_diagnostics["topic_count"].clone(), + "subscriber_count": topic_diagnostics["subscriber_count"].clone(), + "orphan_subscriber_count": topic_diagnostics["orphan_subscriber_count"].clone(), + "duplicate_normalized_subscription_count": topic_diagnostics["duplicate_normalized_subscription_count"].clone(), + "missing_subscriber_identity_count": topic_diagnostics["missing_subscriber_identity_count"].clone(), + "identity_announce_update_supported": topic_diagnostics["identity_announce_update_supported"].clone(), + "last_identity_announce_update_error": topic_diagnostics["last_identity_announce_update_error"].clone(), + "topics": topic_diagnostics, "services": runtime_services_payload(&state), "outbound_delivery": outbound_delivery, "lxmf_sdk": lxmf_sdk_diagnostics_payload(state, &outbound_delivery)?, @@ -26137,52 +26150,6 @@ fn validate_ini_text(config_text: &str) -> Vec { errors } -async fn apply_config_text( - State(state): State, - body: String, -) -> Result, ApiError> { - let display_name = config_text_section_value(&body, "hub", &["display_name"]) - .unwrap_or_else(|| "RCH".to_string()); - let response = apply_config_file(state.config_path.clone(), body, ConfigFileKind::Hub)?; - if let Some(data_plane) = state.lxmf_zmq_data_plane.clone() { - let identity = tokio::task::spawn_blocking(move || { - data_plane.update_identity_announce( - display_name, - vec![ - "r3akt".to_string(), - "emergencymessages".to_string(), - "telemetry".to_string(), - ], - BTreeMap::from([("service".to_string(), Value::String("rch".to_string()))]), - ) - }) - .await - .map_err(|error| { - ApiError::ServiceUnavailable(format!( - "RCH identity announce update task failed: {error}" - )) - })? - .map_err(|error| { - ApiError::ServiceUnavailable(format!("RCH identity announce update failed: {error}")) - })?; - let expected_source = state - .reticulumd_source - .as_deref() - .map(String::as_str) - .unwrap_or(""); - if identity - .delivery_destination - .as_deref() - .is_none_or(|destination| !destination.eq_ignore_ascii_case(expected_source)) - { - return Err(ApiError::ServiceUnavailable( - "RCH identity update returned an unexpected delivery destination".to_string(), - )); - } - } - Ok(response) -} - async fn apply_reticulum_config_text( State(state): State, body: String, @@ -27129,6 +27096,14 @@ async fn delete_topic( .ok_or_else(|| ApiError::NotFound(format!("Topic not found: {topic_id}")))?; drop(topics); delete_topic_row(&state, &topic.topic_id)?; + let mut subscribers = state + .subscribers + .write() + .map_err(|error| ApiError::Internal(error.to_string()))?; + subscribers.retain(|_, subscriber| { + normalize_topic_id(Some(&subscriber.topic_id)).as_deref() != Some(topic.topic_id.as_str()) + }); + drop(subscribers); clear_attachment_topic_links(&state, &topic.topic_id)?; Ok(Json(topic)) } @@ -34836,8 +34811,7 @@ mod tests { .expect("body") .to_bytes(); let payload: serde_json::Value = serde_json::from_slice(&body).expect("json"); - assert_eq!(payload.as_array().expect("subscribers").len(), 1); - assert_eq!(payload[0]["TopicID"], "ops"); + assert!(payload.as_array().expect("subscribers").is_empty()); } #[tokio::test] @@ -46516,6 +46490,14 @@ mod tests { assert_eq!(payload["shutdown_requested"], false); assert_eq!(payload["reticulumd_rpc_configured"], false); assert_eq!(payload["reticulumd_source_configured"], false); + assert_eq!(payload["topic_count"], 0); + assert_eq!(payload["subscriber_count"], 0); + assert_eq!(payload["orphan_subscriber_count"], 0); + assert_eq!(payload["duplicate_normalized_subscription_count"], 0); + assert_eq!(payload["missing_subscriber_identity_count"], 0); + assert_eq!(payload["identity_announce_update_supported"], false); + assert!(payload["last_identity_announce_update_error"].is_null()); + assert_eq!(payload["topics"]["count"], 0); let services = payload["services"].as_array().expect("services"); assert_eq!(services.len(), 4); assert_eq!( @@ -46889,7 +46871,7 @@ mod tests { let payload: serde_json::Value = serde_json::from_slice(&body).expect("json"); assert_eq!(payload["persistence"]["configured"], true); assert_eq!(payload["persistence"]["backend"], "sqlite"); - assert_eq!(payload["persistence"]["schema_version"], "2"); + assert_eq!(payload["persistence"]["schema_version"], "3"); assert!( payload["persistence"]["path"] .as_str() @@ -58129,4 +58111,6 @@ mod tests { let _cleanup = std::fs::remove_dir_all(test_dir); } + + include!("topic_diagnostics_tests.rs"); } diff --git a/crates/r3akt-rch-server/src/topic_diagnostics.rs b/crates/r3akt-rch-server/src/topic_diagnostics.rs new file mode 100644 index 0000000..123e68e --- /dev/null +++ b/crates/r3akt-rch-server/src/topic_diagnostics.rs @@ -0,0 +1,149 @@ +use super::{ + ApiError, AppState, BTreeMap, ConfigFileKind, HashMap, HashSet, Json, State, Value, + apply_config_file, config_text_section_value, identity_announce_matches_destination, json, + load_identity_announces_for_state, normalize_identity_key, normalize_topic_id, +}; + +pub(super) fn topic_subscription_diagnostics(state: &AppState) -> Result { + let topic_ids = state + .topics + .read() + .map_err(|error| ApiError::Internal(error.to_string()))? + .keys() + .filter_map(|topic_id| normalize_topic_id(Some(topic_id))) + .collect::>(); + let subscribers = state + .subscribers + .read() + .map_err(|error| ApiError::Internal(error.to_string()))? + .values() + .cloned() + .collect::>(); + + let topic_count = topic_ids.len(); + let subscriber_count = subscribers.len(); + let mut orphan_subscriber_count = 0_usize; + let mut normalized_pairs = HashMap::<(String, String), usize>::new(); + let mut subscriber_destinations = Vec::with_capacity(subscribers.len()); + let mut missing_subscriber_identity_count = 0_usize; + + for subscriber in &subscribers { + let normalized_topic_id = normalize_topic_id(Some(&subscriber.topic_id)); + if normalized_topic_id + .as_ref() + .is_none_or(|topic_id| !topic_ids.contains(topic_id)) + { + orphan_subscriber_count = orphan_subscriber_count.saturating_add(1); + } + + let Some(destination) = normalize_identity_key(&subscriber.destination) else { + missing_subscriber_identity_count = missing_subscriber_identity_count.saturating_add(1); + continue; + }; + subscriber_destinations.push(destination.clone()); + if let Some(topic_id) = normalized_topic_id { + *normalized_pairs.entry((destination, topic_id)).or_default() += 1; + } + } + + let duplicate_normalized_subscription_count = normalized_pairs + .values() + .map(|count| count.saturating_sub(1)) + .sum::(); + let announces = load_identity_announces_for_state(state)?; + missing_subscriber_identity_count = missing_subscriber_identity_count.saturating_add( + subscriber_destinations + .iter() + .filter(|destination| { + !announces + .iter() + .any(|record| identity_announce_matches_destination(record, destination)) + }) + .count(), + ); + let last_identity_announce_update_error = state + .identity_announce_update_error + .read() + .map_err(|error| ApiError::Internal(error.to_string()))? + .clone(); + let identity_announce_update_supported = state.lxmf_zmq_data_plane.is_some(); + + Ok(json!({ + "topic_count": topic_count, + "subscriber_count": subscriber_count, + "orphan_subscriber_count": orphan_subscriber_count, + "duplicate_normalized_subscription_count": duplicate_normalized_subscription_count, + "missing_subscriber_identity_count": missing_subscriber_identity_count, + "identity_announce_update_supported": identity_announce_update_supported, + "last_identity_announce_update_error": last_identity_announce_update_error, + "count": topic_count, + "subscribers": subscriber_count, + "orphan_subscribers": orphan_subscriber_count, + "duplicate_normalized_subscriptions": duplicate_normalized_subscription_count, + "missing_subscriber_identities": missing_subscriber_identity_count, + })) +} + +pub(super) fn record_identity_announce_update_error( + state: &AppState, + error: Option, +) -> Result<(), ApiError> { + *state + .identity_announce_update_error + .write() + .map_err(|poisoned| ApiError::Internal(poisoned.to_string()))? = error; + Ok(()) +} + +pub(super) async fn apply_config_text( + State(state): State, + body: String, +) -> Result, ApiError> { + let display_name = config_text_section_value(&body, "hub", &["display_name"]) + .unwrap_or_else(|| "RCH".to_string()); + let response = apply_config_file(state.config_path.clone(), body, ConfigFileKind::Hub)?; + if let Some(data_plane) = state.lxmf_zmq_data_plane.clone() { + let identity = match tokio::task::spawn_blocking(move || { + data_plane.update_identity_announce( + display_name, + vec![ + "r3akt".to_string(), + "emergencymessages".to_string(), + "telemetry".to_string(), + ], + BTreeMap::from([("service".to_string(), Value::String("rch".to_string()))]), + ) + }) + .await + { + Ok(Ok(identity)) => identity, + Ok(Err(error)) => { + let message = format!("RCH identity announce update failed: {error}"); + record_identity_announce_update_error(&state, Some(message.clone()))?; + return Err(ApiError::ServiceUnavailable(message)); + } + Err(error) => { + let message = format!("RCH identity announce update task failed: {error}"); + record_identity_announce_update_error(&state, Some(message.clone()))?; + return Err(ApiError::ServiceUnavailable(message)); + } + }; + let expected_source = state + .reticulumd_source + .as_deref() + .map(String::as_str) + .unwrap_or(""); + if identity + .delivery_destination + .as_deref() + .is_none_or(|destination| !destination.eq_ignore_ascii_case(expected_source)) + { + let message = + "RCH identity update returned an unexpected delivery destination".to_string(); + record_identity_announce_update_error(&state, Some(message.clone()))?; + return Err(ApiError::ServiceUnavailable(message)); + } + record_identity_announce_update_error(&state, None)?; + } + Ok(response) +} diff --git a/crates/r3akt-rch-server/src/topic_diagnostics_tests.rs b/crates/r3akt-rch-server/src/topic_diagnostics_tests.rs new file mode 100644 index 0000000..6d7a1f1 --- /dev/null +++ b/crates/r3akt-rch-server/src/topic_diagnostics_tests.rs @@ -0,0 +1,64 @@ +#[test] +fn runtime_diagnostics_reports_topic_subscription_invariants() { + let state = crate::AppState::default(); + state.topics.write().expect("topics").insert( + "ops".to_string(), + crate::TopicRecord { + topic_id: "ops".to_string(), + topic_name: "Ops".to_string(), + topic_path: "ops".to_string(), + topic_description: String::new(), + }, + ); + let mut subscribers = state.subscribers.write().expect("subscribers"); + subscribers.insert( + "one".to_string(), + crate::SubscriberRecord { + subscriber_id: "one".to_string(), + destination: "AABB".to_string(), + topic_id: "ops".to_string(), + reject_tests: None, + metadata: json!({}), + }, + ); + subscribers.insert( + "two".to_string(), + crate::SubscriberRecord { + subscriber_id: "two".to_string(), + destination: " aabb ".to_string(), + topic_id: "ops".to_string(), + reject_tests: None, + metadata: json!({}), + }, + ); + subscribers.insert( + "three".to_string(), + crate::SubscriberRecord { + subscriber_id: "three".to_string(), + destination: String::new(), + topic_id: "missing".to_string(), + reject_tests: None, + metadata: json!({}), + }, + ); + drop(subscribers); + + let diagnostics = crate::runtime_diagnostics_payload(&state).expect("diagnostics"); + assert_eq!(diagnostics["topic_count"], 1); + assert_eq!(diagnostics["subscriber_count"], 3); + assert_eq!(diagnostics["orphan_subscriber_count"], 1); + assert_eq!(diagnostics["duplicate_normalized_subscription_count"], 1); + assert_eq!(diagnostics["missing_subscriber_identity_count"], 3); + assert_eq!(diagnostics["topics"]["count"], 1); + + crate::topic_diagnostics::record_identity_announce_update_error( + &state, + Some("identity update failed".to_string()), + ) + .expect("record identity error"); + let diagnostics = crate::runtime_diagnostics_payload(&state).expect("diagnostics"); + assert_eq!( + diagnostics["last_identity_announce_update_error"], + "identity update failed" + ); +} diff --git a/crates/r3akt-transport-rns/src/identity_update_tests.rs b/crates/r3akt-transport-rns/src/identity_update_tests.rs new file mode 100644 index 0000000..6da3c35 --- /dev/null +++ b/crates/r3akt-transport-rns/src/identity_update_tests.rs @@ -0,0 +1,168 @@ +#[test] +fn zmq_data_plane_registers_and_updates_independent_rch_identity() { + let (command_endpoint, response_endpoint) = unused_zmq_endpoint_pair_v4(); + let captured = Arc::new(Mutex::new(Vec::new())); + let identity_result = serde_json::json!({ + "identity": { + "identity": "11111111111111111111111111111111", + "delivery_destination": "22222222222222222222222222222222", + "public_key": "public-key", + "display_name": "RCH", + "capabilities": ["r3akt"], + "metadata": {"service": "rch"}, + "extensions": {} + } + }); + let server = spawn_zmq_sequence_server( + command_endpoint.clone(), + vec![ + serde_json::json!({ + "runtime_id": "runtime-rch-zmq", + "effective_capabilities": [ + "sdk.capability.identity_multi", + "sdk.capability.identity_import_export", + "sdk.capability.identity_discovery" + ] + }), + identity_result.clone(), + serde_json::json!({"accepted": true}), + serde_json::json!({ + "accepted": true, + "identity": "11111111111111111111111111111111", + "delivery_destination": "22222222222222222222222222222222" + }), + identity_result, + serde_json::json!({"accepted": true}), + serde_json::json!({ + "accepted": true, + "identity": "11111111111111111111111111111111", + "delivery_destination": "22222222222222222222222222222222" + }), + ], + Arc::clone(&captured), + ); + let data_plane = + ZmqDataPlane::new(command_endpoint, response_endpoint).expect("data plane"); + let registered = data_plane + .register_identity(RchServiceIdentityConfig { + private_key: vec![7_u8; 64], + display_name: "RCH".to_string(), + capabilities: vec!["r3akt".to_string()], + metadata: BTreeMap::from([("service".to_string(), serde_json::json!("rch"))]), + }) + .expect("register identity"); + let updated = data_plane + .update_identity_announce( + "Field RCH", + vec!["r3akt".to_string(), "telemetry".to_string()], + BTreeMap::from([ + ("service".to_string(), serde_json::json!("rch")), + ("revision".to_string(), serde_json::json!(2)), + ]), + ) + .expect("update identity"); + data_plane.shutdown().expect("shutdown"); + server.join().expect("server joined"); + + assert_eq!(registered.identity.0, updated.identity.0); + assert_eq!( + registered.delivery_destination, + updated.delivery_destination + ); + let captured = captured.lock().expect("captured requests"); + assert_eq!( + captured + .iter() + .map(|request| request.method.as_str()) + .collect::>(), + vec![ + "sdk_negotiate_v2", + "sdk_identity_import_v2", + "sdk_identity_activate_v2", + "sdk_identity_announce_now_v2", + "sdk_identity_import_v2", + "sdk_identity_activate_v2", + "sdk_identity_announce_now_v2", + ] + ); + assert_eq!(captured[1].params["display_name"], "RCH"); + assert_eq!(captured[3].params["display_name"], "RCH"); + assert_eq!(captured[4].params["display_name"], "Field RCH"); + assert_eq!(captured[6].params["display_name"], "Field RCH"); + assert_eq!( + captured[6].params["capabilities"], + serde_json::json!(["r3akt", "telemetry"]) + ); + assert_eq!(captured[4].params["metadata"]["service"], "rch"); + assert_eq!(captured[4].params["metadata"]["revision"], 2); + assert_eq!(captured[6].params["metadata"]["revision"], 2); +} + +#[test] +fn zmq_data_plane_identity_update_maps_sdk_errors() { + let (command_endpoint, response_endpoint) = unused_zmq_endpoint_pair_v4(); + let captured = Arc::new(Mutex::new(Vec::new())); + let identity_result = serde_json::json!({ + "identity": { + "identity": "11111111111111111111111111111111", + "delivery_destination": "22222222222222222222222222222222", + "public_key": "public-key", + "display_name": "RCH", + "capabilities": ["r3akt"], + "metadata": {"service": "rch"}, + "extensions": {} + } + }); + let server = spawn_zmq_sequence_server( + command_endpoint.clone(), + vec![ + serde_json::json!({"runtime_id": "runtime-rch-zmq"}), + identity_result, + serde_json::json!({"accepted": true}), + serde_json::json!({"accepted": true}), + serde_json::json!({ + "__rpc_error": { + "code": "SDK_IDENTITY_ANNOUNCE_FAILED", + "message": "identity import rejected", + "machine_code": "SDK_IDENTITY_ANNOUNCE_FAILED", + "category": "identity", + "retryable": false + } + }), + ], + Arc::clone(&captured), + ); + let data_plane = + ZmqDataPlane::new(command_endpoint, response_endpoint).expect("data plane"); + data_plane + .register_identity(RchServiceIdentityConfig { + private_key: vec![7_u8; 64], + display_name: "RCH".to_string(), + capabilities: vec!["r3akt".to_string()], + metadata: BTreeMap::new(), + }) + .expect("register identity"); + let error = data_plane + .update_identity_announce("Field RCH", vec!["r3akt".to_string()], BTreeMap::new()) + .expect_err("identity update error"); + data_plane.shutdown().expect("shutdown"); + drop(data_plane); + server.join().expect("server joined"); + + match error { + TransportError::Sdk { + code, + category, + retryable, + message, + } => { + assert_eq!(code, "SDK_IDENTITY_ANNOUNCE_FAILED"); + assert_eq!(category.as_deref(), Some("Internal")); + assert!(!retryable); + assert_eq!(message, "identity import rejected"); + } + other => panic!("expected SDK error, got {other:?}"), + } + let captured = captured.lock().expect("captured requests"); + assert_eq!(captured[4].method, "sdk_identity_import_v2"); +} diff --git a/crates/r3akt-transport-rns/src/lib.rs b/crates/r3akt-transport-rns/src/lib.rs index a367b81..5b2c4fe 100644 --- a/crates/r3akt-transport-rns/src/lib.rs +++ b/crates/r3akt-transport-rns/src/lib.rs @@ -1366,6 +1366,7 @@ struct ZmqSdkActorRequest { struct ZmqSdkActorSession { client: LxmfSdkClient, runtime_info: ZmqRuntimeInfo, + identity: Option, } impl ZmqDataPlane { @@ -1946,39 +1947,24 @@ fn run_zmq_data_plane_actor( send_actor_response(&request.response, result, "missing session"); continue; }; - let payload = match request.payload { + let identity_after_request = match &request.payload { + ZmqSdkActorPayload::RegisterIdentity(config) => Some(config.clone()), ZmqSdkActorPayload::UpdateIdentity { display_name, capabilities, metadata, - } => { - let Some(mut config) = service_identity.clone() else { - let result = Err(TransportError::Send( - "RCH service identity is not registered".to_string(), - )); - metrics.record_result(&result, response_started.elapsed()); - send_actor_response( - &request.response, - result, - "identity update without registration", - ); - continue; - }; - config.display_name = display_name; - config.capabilities = capabilities; - config.metadata = metadata; - ZmqSdkActorPayload::RegisterIdentity(config) - } - payload => payload, - }; - let registration = match &payload { - ZmqSdkActorPayload::RegisterIdentity(config) => Some(config.clone()), + } => service_identity.clone().map(|mut config| { + config.display_name.clone_from(display_name); + config.capabilities.clone_from(capabilities); + config.metadata.clone_from(metadata); + config + }), _ => None, }; - let result = send_lxmf_zmq_actor_request(active_session, payload); + let result = send_lxmf_zmq_actor_request(active_session, request.payload); if result.is_ok() { - if let Some(registration) = registration { - service_identity = Some(registration); + if let Some(identity) = identity_after_request { + service_identity = Some(identity); } } metrics.record_result(&result, response_started.elapsed()); @@ -2064,6 +2050,7 @@ fn open_zmq_sdk_actor_session( Ok(ZmqSdkActorSession { client, runtime_info, + identity: None, }) } @@ -2113,9 +2100,12 @@ fn send_lxmf_zmq_actor_request( ZmqSdkActorPayload::RegisterIdentity(config) => { register_zmq_actor_identity(session, config).map(ZmqSdkActorResponse::Identity) } - ZmqSdkActorPayload::UpdateIdentity { .. } => Err(TransportError::Send( - "identity update was not resolved by the actor".to_string(), - )), + ZmqSdkActorPayload::UpdateIdentity { + display_name, + capabilities, + metadata, + } => update_zmq_actor_identity(session, display_name, capabilities, metadata) + .map(ZmqSdkActorResponse::Identity), ZmqSdkActorPayload::Announce => { send_lxmf_zmq_actor_announce(session).map(ZmqSdkActorResponse::Announce) } @@ -2135,11 +2125,37 @@ fn send_lxmf_zmq_actor_request( fn register_zmq_actor_identity( session: &mut ZmqSdkActorSession, config: RchServiceIdentityConfig, +) -> Result { + let identity = announce_zmq_actor_identity(session, &config)?; + session.identity = Some(config); + Ok(identity) +} + +fn update_zmq_actor_identity( + session: &mut ZmqSdkActorSession, + display_name: String, + capabilities: Vec, + metadata: BTreeMap, +) -> Result { + let Some(mut config) = session.identity.clone() else { + return Err(TransportError::Send( + "RCH service identity is not registered".to_string(), + )); + }; + config.display_name = display_name; + config.capabilities = capabilities; + config.metadata = metadata; + register_zmq_actor_identity(session, config) +} + +fn announce_zmq_actor_identity( + session: &mut ZmqSdkActorSession, + config: &RchServiceIdentityConfig, ) -> Result { let identity = LxmfSdkIdentity::identity_import( &session.client, IdentityImportRequest { - bundle_base64: base64::engine::general_purpose::STANDARD.encode(config.private_key), + bundle_base64: base64::engine::general_purpose::STANDARD.encode(&config.private_key), passphrase: None, display_name: Some(config.display_name.clone()), capabilities: config.capabilities.clone(), @@ -2154,9 +2170,9 @@ fn register_zmq_actor_identity( &session.client, IdentityAnnounceRequest { identity: Some(IdentityRef(identity.identity.0.clone())), - display_name: Some(config.display_name), - capabilities: config.capabilities, - metadata: config.metadata, + display_name: Some(config.display_name.clone()), + capabilities: config.capabilities.clone(), + metadata: config.metadata.clone(), extensions: BTreeMap::new(), }, ) @@ -3994,10 +4010,21 @@ mod tests { } else { response }; - let rpc_response = ReticulumdRpcResponse { - id: envelope.request_id, - result: Some(response), - error: None, + let rpc_response = if let Some(error) = response.get("__rpc_error") { + ReticulumdRpcResponse { + id: envelope.request_id, + result: None, + error: Some( + serde_json::from_value(error.clone()) + .expect("test rpc error response"), + ), + } + } else { + ReticulumdRpcResponse { + id: envelope.request_id, + result: Some(response), + error: None, + } }; let response_payload = encode_frame(&rpc_response).expect("encode response"); let mut response_socket = PushSocket::new(); @@ -4991,100 +5018,6 @@ mod tests { assert_eq!(captured[1].params, serde_json::json!({})); } - #[test] - fn zmq_data_plane_registers_and_updates_independent_rch_identity() { - let (command_endpoint, response_endpoint) = unused_zmq_endpoint_pair_v4(); - let captured = Arc::new(Mutex::new(Vec::new())); - let identity_result = serde_json::json!({ - "identity": { - "identity": "11111111111111111111111111111111", - "delivery_destination": "22222222222222222222222222222222", - "public_key": "public-key", - "display_name": "RCH", - "capabilities": ["r3akt"], - "metadata": {"service": "rch"}, - "extensions": {} - } - }); - let server = spawn_zmq_sequence_server( - command_endpoint.clone(), - vec![ - serde_json::json!({ - "runtime_id": "runtime-rch-zmq", - "effective_capabilities": [ - "sdk.capability.identity_multi", - "sdk.capability.identity_import_export", - "sdk.capability.identity_discovery" - ] - }), - identity_result.clone(), - serde_json::json!({"accepted": true}), - serde_json::json!({ - "accepted": true, - "identity": "11111111111111111111111111111111", - "delivery_destination": "22222222222222222222222222222222" - }), - identity_result, - serde_json::json!({"accepted": true}), - serde_json::json!({ - "accepted": true, - "identity": "11111111111111111111111111111111", - "delivery_destination": "22222222222222222222222222222222" - }), - ], - Arc::clone(&captured), - ); - let data_plane = - ZmqDataPlane::new(command_endpoint, response_endpoint).expect("data plane"); - let registered = data_plane - .register_identity(RchServiceIdentityConfig { - private_key: vec![7_u8; 64], - display_name: "RCH".to_string(), - capabilities: vec!["r3akt".to_string()], - metadata: BTreeMap::from([("service".to_string(), serde_json::json!("rch"))]), - }) - .expect("register identity"); - let updated = data_plane - .update_identity_announce( - "Field RCH", - vec!["r3akt".to_string(), "telemetry".to_string()], - BTreeMap::from([("service".to_string(), serde_json::json!("rch"))]), - ) - .expect("update identity"); - data_plane.shutdown().expect("shutdown"); - server.join().expect("server joined"); - - assert_eq!(registered.identity.0, updated.identity.0); - assert_eq!( - registered.delivery_destination, - updated.delivery_destination - ); - let captured = captured.lock().expect("captured requests"); - assert_eq!( - captured - .iter() - .map(|request| request.method.as_str()) - .collect::>(), - vec![ - "sdk_negotiate_v2", - "sdk_identity_import_v2", - "sdk_identity_activate_v2", - "sdk_identity_announce_now_v2", - "sdk_identity_import_v2", - "sdk_identity_activate_v2", - "sdk_identity_announce_now_v2", - ] - ); - assert_eq!(captured[1].params["display_name"], "RCH"); - assert_eq!(captured[3].params["display_name"], "RCH"); - assert_eq!(captured[4].params["display_name"], "Field RCH"); - assert_eq!(captured[6].params["display_name"], "Field RCH"); - assert_eq!( - captured[6].params["capabilities"], - serde_json::json!(["r3akt", "telemetry"]) - ); - } - #[test] fn delivery_status_parser_accepts_typed_sdk_snapshot() { let snapshot = delivery_snapshot_from_status_result( @@ -5860,6 +5793,8 @@ mod tests { assert_eq!(received.topic.as_str(), "r3akt-live-reticulumd"); assert_eq!(received.source.as_str(), expected_source.as_str()); } + + include!("identity_update_tests.rs"); } #[cfg(test)]