fix: correct RCH topic subscriptions and diagnostics

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);

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);
+}

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<Arc<String>>,
     lxmf_zmq_data_plane: Option<Arc<ZmqDataPlane>>,
     rch_service_identity_config: Option<Arc<RchServiceIdentityConfig>>,
+    identity_announce_update_error: Arc<RwLock<Option<String>>>,
     rch_source_assertion: Option<Arc<String>>,
     runtime_control: Arc<RwLock<RuntimeControlState>>,
     managed_reticulumd: Arc<RwLock<ManagedReticulumdState>>,
@@ -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<Value, ApiError> {
         .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<Value, ApiError> {
         "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<String> {
     errors
 }
 
-async fn apply_config_text(
-    State(state): State<AppState>,
-    body: String,
-) -> Result<Json<Value>, 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<AppState>,
     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");
 }