From 177b9a3b394bb63a71510215be4fe4f91c3604bb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=5FMr=E4=B8=B6s=20=E3=82=9B=F0=9F=98=81=C2=AE?= <31917567+ma123456mai@users.noreply.github.com> Date: Mon, 12 Oct 2026 19:50:43 +0800 Subject: [PATCH] fix(mqtt): preserve console history across disconnects --- .../src/components/mqtt/MqttAdminConsole.vue | 5 +- .../src/components/mqtt/MqttPublishDialog.vue | 100 ++++++++- crates/dbx-core/src/admin/mqtt/client.rs | 26 ++- crates/dbx-core/src/admin/mqtt/service.rs | 2 +- crates/dbx-core/src/connection/mod.rs | 81 +++++++- crates/dbx-core/src/persistence/storage.rs | 191 +++++++++++++++++ src-tauri/src/commands/mqtt_cmd.rs | 196 +++++++++++++++++- 7 files changed, 587 insertions(+), 14 deletions(-) diff --git a/apps/desktop/src/components/mqtt/MqttAdminConsole.vue b/apps/desktop/src/components/mqtt/MqttAdminConsole.vue index 98e4605358..6bade005d8 100644 --- a/apps/desktop/src/components/mqtt/MqttAdminConsole.vue +++ b/apps/desktop/src/components/mqtt/MqttAdminConsole.vue @@ -64,6 +64,7 @@ const formNoLocal = ref(restoredUiState.formNoLocal ?? false); const formEnabled = ref(restoredUiState.formEnabled ?? true); const editingTopic = ref(restoredUiState.editingTopic ?? null); const MQTT_PUBLISH_PANEL_COLLAPSED_STORAGE_KEY = "dbx-mqtt-publish-panel-collapsed"; +const MQTT_MESSAGE_HISTORY_LIMIT = 400; const publishPanelCollapsed = ref(restoredUiState.publishPanelCollapsed ?? safeLocalStorageGet(MQTT_PUBLISH_PANEL_COLLAPSED_STORAGE_KEY) === "true"); trackUiState(() => ({ @@ -150,7 +151,7 @@ async function refreshData() { mqttGetBrokerInfo(props.connectionId) as Promise, mqttListTopics(props.connectionId) as Promise<[string, string][]>, mqttListSavedTopicConfigs(props.connectionId) as Promise, - mqttGetMessages(props.connectionId, selectedTopic.value || undefined, 50) as Promise, + mqttGetMessages(props.connectionId, selectedTopic.value || undefined, MQTT_MESSAGE_HISTORY_LIMIT) as Promise, ]); brokerInfo.value = info; subscribedTopics.value = active; @@ -305,7 +306,7 @@ function startPolling() { pollingTimer.value = setInterval(async () => { if (messagesPaused.value) return; try { - const nextMessages = (await mqttGetMessages(props.connectionId, selectedTopic.value || undefined, 50)) as MqttMessage[]; + const nextMessages = (await mqttGetMessages(props.connectionId, selectedTopic.value || undefined, MQTT_MESSAGE_HISTORY_LIMIT)) as MqttMessage[]; if (!messagesPaused.value) messages.value = nextMessages; } catch { /* ignore polling errors */ diff --git a/apps/desktop/src/components/mqtt/MqttPublishDialog.vue b/apps/desktop/src/components/mqtt/MqttPublishDialog.vue index 0f7c53b3c6..a9e1c1e8ce 100644 --- a/apps/desktop/src/components/mqtt/MqttPublishDialog.vue +++ b/apps/desktop/src/components/mqtt/MqttPublishDialog.vue @@ -28,11 +28,13 @@ const publishPanelRef = ref(null); const payloadTextareaRef = ref(null); const payloadHeight = ref(80); const resizing = ref(false); +const formVersion = ref(0); const MQTT_PAYLOAD_MIN_HEIGHT_PX = 60; const MQTT_PAYLOAD_MAX_PANEL_RATIO = 0.5; const MQTT_PAYLOAD_KEYBOARD_STEP_PX = 10; const MQTT_PAYLOAD_HEIGHT_STORAGE_KEY = "dbx-mqtt-payload-height"; +const MQTT_LAST_PUBLISHED_MESSAGE_KEY_PREFIX = "dbx-mqtt-last-published-message:"; let resizeStartY = 0; let resizeStartHeight = 0; let panelResizeObserver: ResizeObserver | undefined; @@ -40,6 +42,14 @@ let panelResizeObserver: ResizeObserver | undefined; const qosLabels = ["QoS 0", "QoS 1", "QoS 2"]; const qosHints = computed(() => [t("connection.mqttQosAtMostOnce"), t("connection.mqttQosAtLeastOnce"), t("connection.mqttQosExactlyOnce")]); +watch( + () => [topic.value, qos.value, retain.value, payloadText.value, encoding.value], + () => { + formVersion.value += 1; + }, + { flush: "sync" }, +); + /* 根据编码格式调整输入提示和占位符 */ const payloadPlaceholder = computed(() => { switch (encoding.value) { @@ -61,12 +71,92 @@ const payloadPlaceholder = computed(() => { watch( () => props.initialTopic, (val) => { - if (val !== undefined && val !== topic.value) { + if (val !== undefined && val !== topic.value && !hasStoredPublishForm()) { topic.value = val; } }, ); +watch( + () => props.connectionId, + () => { + restoreLastPublishedMessage(); + }, +); + +function lastPublishedMessageStorageKey() { + return `${MQTT_LAST_PUBLISHED_MESSAGE_KEY_PREFIX}${encodeURIComponent(props.connectionId)}`; +} + +function hasStoredPublishForm() { + try { + return localStorage.getItem(lastPublishedMessageStorageKey()) !== null; + } catch { + return false; + } +} + +function restoreLastPublishedMessage(): boolean { + try { + const raw = localStorage.getItem(lastPublishedMessageStorageKey()); + if (!raw) { + topic.value = props.initialTopic || ""; + qos.value = 0; + retain.value = false; + payloadText.value = ""; + encoding.value = "plaintext"; + return false; + } + const saved = JSON.parse(raw) as Partial<{ + cleared: boolean; + topic: string; + qos: 0 | 1 | 2; + retain: boolean; + payloadText: string; + encoding: PayloadEncoding; + }>; + if (saved.cleared === true) { + topic.value = ""; + qos.value = 0; + retain.value = false; + payloadText.value = ""; + encoding.value = "plaintext"; + return true; + } + if (typeof saved.topic === "string") topic.value = saved.topic; + if (saved.qos === 0 || saved.qos === 1 || saved.qos === 2) qos.value = saved.qos; + if (typeof saved.retain === "boolean") retain.value = saved.retain; + if (typeof saved.payloadText === "string") payloadText.value = saved.payloadText; + if (saved.encoding && PAYLOAD_ENCODINGS.includes(saved.encoding)) encoding.value = saved.encoding; + return true; + } catch { + // Keep the default form usable if local browser storage is unavailable or invalid. + topic.value = props.initialTopic || ""; + qos.value = 0; + retain.value = false; + payloadText.value = ""; + encoding.value = "plaintext"; + return false; + } +} + +function rememberLastPublishedMessage() { + try { + localStorage.setItem( + lastPublishedMessageStorageKey(), + JSON.stringify({ + topic: topic.value.trim(), + qos: qos.value, + retain: retain.value, + payloadText: payloadText.value, + encoding: encoding.value, + }), + ); + } catch { + // The publish itself has succeeded; a storage failure must not report it as failed. + } +} + /** 计算消息输入框允许使用的最大高度，避免挤占全部消息列表区域 */ function maxPayloadHeight() { const containerHeight = publishPanelRef.value?.parentElement?.clientHeight || window.innerHeight || 0; @@ -129,6 +219,7 @@ function stopResize() { } onMounted(() => { + restoreLastPublishedMessage(); const savedHeight = Number.parseInt(localStorage.getItem(MQTT_PAYLOAD_HEIGHT_STORAGE_KEY) ?? "", 10); if (Number.isFinite(savedHeight)) payloadHeight.value = clampPayloadHeight(savedHeight); @@ -173,6 +264,7 @@ async function publish() { retain: retain.value, }); success.value = true; + rememberLastPublishedMessage(); emit("published"); /* 成功后保留表单内容便于连续测试，1.5 秒后清除成功提示 */ setTimeout(() => { @@ -193,6 +285,12 @@ function clearForm() { encoding.value = "plaintext"; error.value = null; success.value = false; + try { + localStorage.setItem(lastPublishedMessageStorageKey(), JSON.stringify({ cleared: true })); + } catch { + // The form is still cleared for this session if browser storage is unavailable. + } + formVersion.value += 1; } diff --git a/crates/dbx-core/src/admin/mqtt/client.rs b/crates/dbx-core/src/admin/mqtt/client.rs index d54a8a72e3..eb054bb374 100644 --- a/crates/dbx-core/src/admin/mqtt/client.rs +++ b/crates/dbx-core/src/admin/mqtt/client.rs @@ -326,6 +326,8 @@ pub struct MqttClient { publish_send_lock: Mutex<()>, /// 最近接收的消息缓冲区（保留最近 N 条） message_buffer: RwLock>, + /// 串行化消息缓冲区与持久化历史之间的复合操作。 + message_history_operation: Mutex<()>, /// 最大缓冲消息数 max_buffer_size: usize, /// 当前订阅生命周期内已见过的保留消息指纹 @@ -419,6 +421,7 @@ impl MqttClient { publish_requests: Mutex::new(PublishRequestTracker::default()), publish_send_lock: Mutex::new(()), message_buffer: RwLock::new(Vec::new()), + message_history_operation: Mutex::new(()), max_buffer_size: 200, seen_retained: RwLock::new(RetainedDedup::new(RETAINED_DEDUP_CAPACITY)), event_loop_handle: RwLock::new(None), @@ -1052,7 +1055,7 @@ impl MqttClient { } /// 发布消息；QoS 0 等待写入网络，QoS 1/2 分别等待 PUBACK/PUBCOMP。 - pub async fn publish(&self, request: &MqttPublishRequest) -> Result<(), String> { + pub async fn publish(&self, request: &MqttPublishRequest) -> Result { validate_publish_topic(&request.topic)?; let payload = if let Some(ref text) = request.payload_text { text.as_bytes().to_vec() @@ -1088,6 +1091,7 @@ impl MqttClient { received_at_ms: now_ms, direction: MqttMessageDirection::Sent, }; + let published_message = sent_msg.clone(); let _send_guard = self.publish_send_lock.lock().await; let (completion, result) = oneshot::channel(); @@ -1097,7 +1101,8 @@ impl MqttClient { return Err(error); } drop(_send_guard); - result.await.map_err(|_| "发布确认通道已关闭，请检查 MQTT 连接状态".to_string())? + result.await.map_err(|_| "发布确认通道已关闭，请检查 MQTT 连接状态".to_string())??; + Ok(published_message) } /// 获取已订阅的 topic 列表 @@ -1130,12 +1135,29 @@ impl MqttClient { filtered } + pub async fn lock_message_history(&self) -> tokio::sync::MutexGuard<'_, ()> { + self.message_history_operation.lock().await + } + /// 清空消息缓冲区 pub async fn clear_messages(&self) { self.message_buffer.write().await.clear(); self.seen_retained.write().await.clear(); } + #[cfg(any(test, feature = "test-support"))] + #[doc(hidden)] + pub async fn test_client_with_messages(messages: Vec) -> Arc { + let options = MqttOptionsV4::new("dbx-mqtt-test", "127.0.0.1", 1883); + let (backend, event_loop) = AsyncClientV4::new(options, 10); + drop(event_loop); + let shutdown_notify = Arc::new(Notify::new()); + let client = + Self::new_with_backend(MqttConnectionConfig::default(), MqttBackendClient::V4(backend), shutdown_notify); + *client.message_buffer.write().await = messages; + client + } + /// 从 topic 列表构建 topic 树 pub async fn build_topic_tree(&self) -> MqttTopicNode { let subscriptions = self.subscriptions.read().await; diff --git a/crates/dbx-core/src/admin/mqtt/service.rs b/crates/dbx-core/src/admin/mqtt/service.rs index 7a26e33f3c..f8750b93bd 100644 --- a/crates/dbx-core/src/admin/mqtt/service.rs +++ b/crates/dbx-core/src/admin/mqtt/service.rs @@ -41,7 +41,7 @@ pub async fn unsubscribe(client: &Arc, topic: &str) -> Result<(), St } /// 发布消息 -pub async fn publish(client: &Arc, request: &MqttPublishRequest) -> Result<(), String> { +pub async fn publish(client: &Arc, request: &MqttPublishRequest) -> Result { client.publish(request).await } diff --git a/crates/dbx-core/src/connection/mod.rs b/crates/dbx-core/src/connection/mod.rs index ca61dee6da..52ca966825 100644 --- a/crates/dbx-core/src/connection/mod.rs +++ b/crates/dbx-core/src/connection/mod.rs @@ -2481,7 +2481,9 @@ impl AppState { pub async fn shutdown(&self, deadline: Duration) { self.invalidate_all_connection_lifecycles(); self.running_queries.cancel_all(); - let removed_pools = self.drain_all_connection_pools().await; + let mut removed_pools = self.drain_all_connection_pools().await; + #[cfg(feature = "mq-admin")] + self.persist_removed_mqtt_history(&mut removed_pools).await; self.transaction_sessions.write().await.clear(); let shutdown = async { @@ -5953,8 +5955,10 @@ impl AppState { pub async fn remove_connection_pools(&self, connection_id: &str) { self.invalidate_connection_lifecycle(connection_id); self.rollback_manual_transaction_sessions(connection_id).await; - let removed = self.drain_connection_pools(connection_id).await; + let mut removed = self.drain_connection_pools(connection_id).await; self.clear_metadata_gates_for_connection(connection_id).await; + #[cfg(feature = "mq-admin")] + self.persist_removed_mqtt_history(&mut removed).await; self.pool_routing_control().close_removed(removed).await; } @@ -5977,11 +5981,32 @@ impl AppState { pub async fn remove_connection_pools_detached(&self, connection_id: &str) { self.invalidate_connection_lifecycle(connection_id); self.rollback_manual_transaction_sessions(connection_id).await; - let removed = self.drain_connection_pools(connection_id).await; + let mut removed = self.drain_connection_pools(connection_id).await; self.clear_metadata_gates_for_connection(connection_id).await; + #[cfg(feature = "mq-admin")] + self.persist_removed_mqtt_history(&mut removed).await; self.pool_routing_control().close_removed_in_background(removed); } + #[cfg(feature = "mq-admin")] + async fn persist_removed_mqtt_history(&self, removed: &mut Vec<(String, PoolKind)>) { + let mut remaining = Vec::with_capacity(removed.len()); + for (pool_key, pool) in std::mem::take(removed) { + match pool { + PoolKind::Mqtt(client) => { + let _history_guard = client.lock_message_history().await; + client.disconnect().await; + let messages = client.get_messages(None, usize::MAX).await; + if let Err(error) = self.storage.reconcile_mqtt_message_history(&pool_key, &messages).await { + log::warn!("Failed to persist MQTT history while closing '{pool_key}': {error}"); + } + } + pool => remaining.push((pool_key, pool)), + } + } + *removed = remaining; + } + /// Close and roll back every manual-transaction session of a connection /// before its pools are drained. The transaction sessions hold dedicated /// connections outside the pools being removed; without this the session @@ -8337,6 +8362,56 @@ mod tests { let _ = std::fs::remove_dir_all(dir); } + #[cfg(feature = "mq-admin")] + #[tokio::test] + async fn mqtt_history_is_persisted_at_every_connection_teardown_boundary() { + use crate::mqtt::client::MqttClient; + use crate::mqtt::types::{MqttMessage, MqttMessageDirection}; + + async fn install_client(state: &AppState, connection_id: &str, timestamp: u64) { + let message = MqttMessage { + topic: format!("devices/{connection_id}"), + payload_base64: "dGFpbA==".to_string(), + payload_text: Some("tail".to_string()), + qos: 1, + retain: false, + received_at_ms: timestamp, + direction: MqttMessageDirection::Received, + }; + let client = MqttClient::test_client_with_messages(vec![message]).await; + state + .update_connection_pools(|connections| { + connections.insert(connection_id.to_string(), PoolKind::Mqtt(client)); + }) + .await; + } + + let (removed_state, removed_dir) = test_app_state().await; + install_client(&removed_state, "removed", 1).await; + removed_state.remove_connection_pools("removed").await; + let removed_history = removed_state.storage.load_mqtt_message_history("removed").await.unwrap(); + assert_eq!(removed_history.len(), 1); + assert_eq!(removed_history[0].topic, "devices/removed"); + + let (detached_state, detached_dir) = test_app_state().await; + install_client(&detached_state, "detached", 2).await; + detached_state.remove_connection_pools_detached("detached").await; + let detached_history = detached_state.storage.load_mqtt_message_history("detached").await.unwrap(); + assert_eq!(detached_history.len(), 1); + assert_eq!(detached_history[0].topic, "devices/detached"); + + let (shutdown_state, shutdown_dir) = test_app_state().await; + install_client(&shutdown_state, "shutdown", 3).await; + shutdown_state.shutdown(Duration::from_millis(100)).await; + let shutdown_history = shutdown_state.storage.load_mqtt_message_history("shutdown").await.unwrap(); + assert_eq!(shutdown_history.len(), 1); + assert_eq!(shutdown_history[0].topic, "devices/shutdown"); + + let _ = std::fs::remove_dir_all(removed_dir); + let _ = std::fs::remove_dir_all(detached_dir); + let _ = std::fs::remove_dir_all(shutdown_dir); + } + #[tokio::test] async fn named_resource_budget_is_shared_across_app_state_views() { let (state, dir) = test_app_state().await; diff --git a/crates/dbx-core/src/persistence/storage.rs b/crates/dbx-core/src/persistence/storage.rs index d4f534fde3..777d07db24 100644 --- a/crates/dbx-core/src/persistence/storage.rs +++ b/crates/dbx-core/src/persistence/storage.rs @@ -16,6 +16,9 @@ use rusqlite::{ use serde::{Deserialize, Serialize}; use uuid::Uuid; +#[cfg(feature = "mq-admin")] +use dbx_types::mqtt::{MqttMessage, MqttMessageDirection}; + use crate::ai::{ AiChatMessage, AiChatSelectionState, AiConfig, AiConfigItem, AiConversation, AiProvider, AiRun, AiRunFifoCategory, AiRunStatus, @@ -83,6 +86,25 @@ const PLUGIN_DATA_GRANTS_KEY: &str = "plugin_data_grants"; const MAX_PLUGIN_DATA_GRANTS_PER_PLUGIN: usize = 512; const APP_STATE_AI_GLOBAL_INSTRUCTIONS_KEY: &str = "ai_global_custom_instructions"; const APP_STATE_AI_CHAT_SELECTION_KEY: &str = "ai_chat_selection_v1"; +// Keep the existing key so previously persisted publish records remain readable. +const MQTT_MESSAGE_HISTORY_STATE_KEY_PREFIX: &str = "mqtt.sent_history."; +#[cfg(feature = "mq-admin")] +const MQTT_MESSAGE_HISTORY_MAX_MESSAGES: usize = 200; +#[cfg(feature = "mq-admin")] +const MQTT_MESSAGE_HISTORY_MAX_BYTES: usize = 8 * 1024 * 1024; + +#[cfg(feature = "mq-admin")] +fn mqtt_history_message_identity(message: &MqttMessage) -> (String, String, Option, u8, bool, bool, u64) { + ( + message.topic.clone(), + message.payload_base64.clone(), + message.payload_text.clone(), + message.qos, + message.retain, + message.direction == MqttMessageDirection::Sent, + message.received_at_ms, + ) +} const SNIPPET_SYNC_IDS_KEY: &str = "snippet_sync_ids"; const SNIPPET_PENDING_CLEANUPS_KEY: &str = "snippet_pending_legacy_cleanups"; const USER_DATA_TABLES: &[&str] = &[ @@ -6952,6 +6974,8 @@ impl Storage { tx.execute("DELETE FROM connections WHERE id = ?1", [id]).map_err(|e| e.to_string())?; tx.execute("DELETE FROM connection_secrets WHERE connection_id = ?1", [id]) .map_err(|e| e.to_string())?; + let mqtt_history_key = format!("{MQTT_MESSAGE_HISTORY_STATE_KEY_PREFIX}{id}"); + tx.execute("DELETE FROM state_store WHERE key = ?1", [mqtt_history_key]).map_err(|e| e.to_string())?; } tx.commit().map_err(|e| e.to_string()) }) @@ -7060,6 +7084,8 @@ impl Storage { if removed { tx.execute("DELETE FROM connection_secrets WHERE connection_id = ?1", [&connection_id]) .map_err(|e| e.to_string())?; + let mqtt_history_key = format!("{MQTT_MESSAGE_HISTORY_STATE_KEY_PREFIX}{connection_id}"); + tx.execute("DELETE FROM state_store WHERE key = ?1", [mqtt_history_key]).map_err(|e| e.to_string())?; } tx.commit().map_err(|e| e.to_string())?; Ok(removed) @@ -7166,6 +7192,125 @@ impl Storage { .await } + /// Append one successfully published message to a connection's bounded + /// local history. The read-modify-write runs in one SQLite transaction so + /// concurrent publishes do not overwrite each other's records. + #[cfg(feature = "mq-admin")] + pub async fn append_mqtt_sent_message(&self, connection_id: &str, message: &MqttMessage) -> Result<(), String> { + if message.direction != MqttMessageDirection::Sent { + return Err("MQTT sent history accepts outgoing messages only".to_string()); + } + self.append_mqtt_history_messages(connection_id, std::slice::from_ref(message), false).await + } + + /// Reconcile live MQTT messages with persisted history without re-adding + /// entries already stored by an earlier publish or console poll. + #[cfg(feature = "mq-admin")] + pub async fn reconcile_mqtt_message_history( + &self, + connection_id: &str, + messages: &[MqttMessage], + ) -> Result<(), String> { + self.append_mqtt_history_messages(connection_id, messages, true).await + } + + #[cfg(feature = "mq-admin")] + async fn append_mqtt_history_messages( + &self, + connection_id: &str, + incoming_messages: &[MqttMessage], + skip_existing: bool, + ) -> Result<(), String> { + if incoming_messages.is_empty() { + return Ok(()); + } + let key = format!("{MQTT_MESSAGE_HISTORY_STATE_KEY_PREFIX}{connection_id}"); + let incoming_messages = incoming_messages.to_vec(); + self.with_conn(move |conn| { + let tx = + conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(|error| error.to_string())?; + let current: Option> = tx + .query_row("SELECT value FROM state_store WHERE key = ?1", [&key], |row| row.get(0)) + .optional() + .map_err(|error| error.to_string())?; + let mut stored_messages: Vec = current + .map(|bytes| serde_json::from_slice(&bytes).map_err(|error| error.to_string())) + .transpose()? + .unwrap_or_default(); + let mut message_counts = HashMap::<(String, String, Option, u8, bool, bool, u64), usize>::new(); + for message in &stored_messages { + *message_counts.entry(mqtt_history_message_identity(message)).or_default() += 1; + } + let mut changed = false; + for message in incoming_messages { + let identity = mqtt_history_message_identity(&message); + if skip_existing { + if let Some(count) = message_counts.get_mut(&identity) { + if *count > 0 { + *count -= 1; + continue; + } + } + *message_counts.entry(identity).or_default() += 1; + } + stored_messages.push(message); + changed = true; + } + if !changed { + tx.commit().map_err(|error| error.to_string())?; + return Ok(()); + } + stored_messages.sort_by_key(|entry| entry.received_at_ms); + if stored_messages.len() > MQTT_MESSAGE_HISTORY_MAX_MESSAGES { + stored_messages.drain(..stored_messages.len() - MQTT_MESSAGE_HISTORY_MAX_MESSAGES); + } + + let bytes = loop { + let bytes = serde_json::to_vec(&stored_messages).map_err(|error| error.to_string())?; + if bytes.len() <= MQTT_MESSAGE_HISTORY_MAX_BYTES { + break bytes; + } + if stored_messages.len() <= 1 { + // Keep the previous history if one message alone is too large. + tx.commit().map_err(|error| error.to_string())?; + return Ok(()); + } + stored_messages.remove(0); + }; + tx.execute( + "INSERT INTO state_store (key, value, content_type, version, payload) \ + VALUES (?1, ?2, 'application/json', 1, x'') \ + ON CONFLICT(key) DO UPDATE SET value = excluded.value, content_type = excluded.content_type, \ + version = version + 1, payload = excluded.payload", + params![key, bytes], + ) + .map_err(|error| error.to_string())?; + tx.commit().map_err(|error| error.to_string()) + }) + .await + } + + /// Load the bounded MQTT message history for one connection. + #[cfg(feature = "mq-admin")] + pub async fn load_mqtt_message_history(&self, connection_id: &str) -> Result, String> { + let key = format!("{MQTT_MESSAGE_HISTORY_STATE_KEY_PREFIX}{connection_id}"); + let Some((bytes, _)) = self.load_state(&key).await? else { + return Ok(Vec::new()); + }; + let mut messages: Vec = serde_json::from_slice(&bytes).map_err(|error| error.to_string())?; + if messages.len() > MQTT_MESSAGE_HISTORY_MAX_MESSAGES { + messages.drain(..messages.len() - MQTT_MESSAGE_HISTORY_MAX_MESSAGES); + } + Ok(messages) + } + + /// Remove MQTT message history when the user clears the console messages. + #[cfg(feature = "mq-admin")] + pub async fn clear_mqtt_message_history(&self, connection_id: &str) -> Result<(), String> { + let key = format!("{MQTT_MESSAGE_HISTORY_STATE_KEY_PREFIX}{connection_id}"); + self.delete_state(&key).await + } + /// Update only the persisted driver identity for an existing connection. /// This is used after runtime driver fallback and deliberately leaves all /// other connection metadata and separately stored secrets untouched. @@ -9633,6 +9778,52 @@ mod tests { use std::collections::BTreeMap; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + #[cfg(feature = "mq-admin")] + #[tokio::test] + async fn mqtt_history_reconciliation_recovers_sent_message_after_initial_append_failure() { + use crate::mqtt::types::{MqttMessage, MqttMessageDirection}; + + let directory = tempfile::tempdir().unwrap(); + let storage = crate::persistence::test_storage::open(&directory.path().join("storage.db")).await.unwrap(); + storage + .with_conn(|connection| { + connection + .execute_batch( + "CREATE TRIGGER fail_mqtt_history_insert \ + BEFORE INSERT ON state_store \ + WHEN NEW.key = 'mqtt.sent_history.retry' \ + BEGIN SELECT RAISE(FAIL, 'injected mqtt history failure'); END;", + ) + .map_err(|error| error.to_string()) + }) + .await + .unwrap(); + let message = MqttMessage { + topic: "devices/retry".to_string(), + payload_base64: "cmV0cnk=".to_string(), + payload_text: Some("retry".to_string()), + qos: 1, + retain: false, + received_at_ms: 42, + direction: MqttMessageDirection::Sent, + }; + + assert!(storage.append_mqtt_sent_message("retry", &message).await.is_err()); + storage + .with_conn(|connection| { + connection.execute_batch("DROP TRIGGER fail_mqtt_history_insert").map_err(|error| error.to_string()) + }) + .await + .unwrap(); + + storage.reconcile_mqtt_message_history("retry", std::slice::from_ref(&message)).await.unwrap(); + storage.reconcile_mqtt_message_history("retry", std::slice::from_ref(&message)).await.unwrap(); + + let history = storage.load_mqtt_message_history("retry").await.unwrap(); + assert_eq!(history.len(), 1); + assert_eq!(super::mqtt_history_message_identity(&history[0]), super::mqtt_history_message_identity(&message)); + } + #[tokio::test] async fn migration_scan_cache_recovers_legacy_success_with_stale_counts() { let dir = temp_data_dir("migration-stale-cache"); diff --git a/src-tauri/src/commands/mqtt_cmd.rs b/src-tauri/src/commands/mqtt_cmd.rs index ef91b23029..b3100b1a66 100644 --- a/src-tauri/src/commands/mqtt_cmd.rs +++ b/src-tauri/src/commands/mqtt_cmd.rs @@ -1,4 +1,6 @@ use serde_json::json; +use std::collections::HashMap; +use std::future::Future; use std::sync::Arc; use tauri::State; @@ -7,6 +9,48 @@ use dbx_core::connection::PoolKind; use dbx_core::mqtt::service; use dbx_core::mqtt::types::*; +const MQTT_LIVE_MESSAGE_LIMIT: usize = 200; +const MQTT_VISIBLE_MESSAGE_LIMIT: usize = MQTT_LIVE_MESSAGE_LIMIT * 2; + +type MqttMessageIdentity = (String, String, Option, u8, bool, bool, u64); + +fn mqtt_message_identity(message: &MqttMessage) -> MqttMessageIdentity { + ( + message.topic.clone(), + message.payload_base64.clone(), + message.payload_text.clone(), + message.qos, + message.retain, + message.direction == MqttMessageDirection::Sent, + message.received_at_ms, + ) +} + +fn mqtt_topic_matches_filter(topic: &str, filter: &str) -> bool { + let topic_segments = topic.split('/').collect::>(); + let filter_segments = filter.split('/').collect::>(); + let (mut topic_index, mut filter_index) = (0, 0); + while filter_index < filter_segments.len() { + match filter_segments[filter_index] { + "#" => return filter_index + 1 == filter_segments.len(), + "+" if topic_index < topic_segments.len() => { + topic_index += 1; + filter_index += 1; + } + segment if topic_index < topic_segments.len() && topic_segments[topic_index] == segment => { + topic_index += 1; + filter_index += 1; + } + _ => return false, + } + } + topic_index == topic_segments.len() +} + +fn mqtt_message_matches_filter(message: &MqttMessage, filter: Option<&str>) -> bool { + filter.is_none_or(|filter| message.topic == filter || mqtt_topic_matches_filter(&message.topic, filter)) +} + /// 从 connections map 中获取 MQTT 客户端 async fn get_mqtt_client( state: &AppState, @@ -108,7 +152,13 @@ pub async fn mqtt_publish( request: MqttPublishRequest, ) -> Result<(), String> { let client = get_mqtt_client(&state, &connection_id).await?; - service::publish(&client, &request).await + let _history_guard = client.lock_message_history().await; + let published_message = service::publish(&client, &request).await?; + + if let Err(error) = state.storage.append_mqtt_sent_message(&connection_id, &published_message).await { + log::warn!("保存 MQTT 发布历史失败: {error}"); + } + Ok(()) } /// 获取已订阅的 topic 列表 @@ -149,13 +199,149 @@ pub async fn mqtt_get_messages( topic_filter: Option, limit: Option, ) -> Result, String> { - let client = get_mqtt_client(&state, &connection_id).await?; - service::get_messages(&client, topic_filter.as_deref(), limit.unwrap_or(50)).await + mqtt_get_messages_inner(&state, connection_id, topic_filter, limit).await +} + +async fn mqtt_get_messages_inner( + state: &AppState, + connection_id: String, + topic_filter: Option, + limit: Option, +) -> Result, String> { + mqtt_get_messages_inner_with_hook(state, connection_id, topic_filter, limit, || async {}).await +} + +async fn mqtt_get_messages_inner_with_hook( + state: &AppState, + connection_id: String, + topic_filter: Option, + limit: Option, + after_snapshot: F, +) -> Result, String> +where + F: FnOnce() -> Fut, + Fut: Future, +{ + let client = get_mqtt_client(state, &connection_id).await?; + let _history_guard = client.lock_message_history().await; + let live_messages = service::get_messages(&client, None, MQTT_LIVE_MESSAGE_LIMIT).await?; + after_snapshot().await; + if let Err(error) = state.storage.reconcile_mqtt_message_history(&connection_id, &live_messages).await { + log::warn!("保存 MQTT 消息历史失败: {error}"); + } + let history = match state.storage.load_mqtt_message_history(&connection_id).await { + Ok(history) => history, + Err(error) => { + log::warn!("读取 MQTT 消息历史失败: {error}"); + Vec::new() + } + }; + let mut history_counts = HashMap::::new(); + for message in &history { + *history_counts.entry(mqtt_message_identity(message)).or_default() += 1; + } + let mut messages = history; + for message in live_messages { + let identity = mqtt_message_identity(&message); + if let Some(count) = history_counts.get_mut(&identity) { + if *count > 0 { + *count -= 1; + continue; + } + } + messages.push(message); + } + messages.sort_by_key(|message| message.received_at_ms); + let limit = limit.unwrap_or(50).min(MQTT_VISIBLE_MESSAGE_LIMIT); + Ok(messages + .into_iter() + .filter(|message| mqtt_message_matches_filter(message, topic_filter.as_deref())) + .rev() + .take(limit) + .collect()) } /// 清空消息历史记录 #[tauri::command] pub async fn mqtt_clear_messages(state: State<'_, Arc>, connection_id: String) -> Result<(), String> { - let client = get_mqtt_client(&state, &connection_id).await?; - service::clear_messages(&client).await + mqtt_clear_messages_inner(&state, connection_id).await +} + +async fn mqtt_clear_messages_inner(state: &AppState, connection_id: String) -> Result<(), String> { + mqtt_clear_messages_inner_with_hook(state, connection_id, || async {}).await +} + +async fn mqtt_clear_messages_inner_with_hook( + state: &AppState, + connection_id: String, + before_lock: F, +) -> Result<(), String> +where + F: FnOnce() -> Fut, + Fut: Future, +{ + let client = get_mqtt_client(state, &connection_id).await?; + before_lock().await; + let _history_guard = client.lock_message_history().await; + service::clear_messages(&client).await?; + state.storage.clear_mqtt_message_history(&connection_id).await +} + +#[cfg(test)] +mod tests { + use super::*; + use dbx_core::mqtt::client::MqttClient; + use tokio::sync::oneshot; + + #[tokio::test] + async fn clear_waits_for_poll_reconciliation_and_prevents_snapshot_writeback() { + let directory = tempfile::tempdir().unwrap(); + let storage = dbx_core::persistence::test_storage::open(&directory.path().join("storage.db")).await.unwrap(); + let state = Arc::new(AppState::new(storage)); + let message = MqttMessage { + topic: "devices/race".to_string(), + payload_base64: "cmFjZQ==".to_string(), + payload_text: Some("race".to_string()), + qos: 1, + retain: false, + received_at_ms: 7, + direction: MqttMessageDirection::Received, + }; + let client = MqttClient::test_client_with_messages(vec![message]).await; + state + .update_connection_pools(|connections| { + connections.insert("mqtt-race".to_string(), PoolKind::Mqtt(client.clone())); + }) + .await; + + let (snapshot_tx, snapshot_rx) = oneshot::channel(); + let (resume_tx, resume_rx) = oneshot::channel(); + let poll_state = Arc::clone(&state); + let poll = tokio::spawn(async move { + mqtt_get_messages_inner_with_hook(&poll_state, "mqtt-race".to_string(), None, None, || async move { + snapshot_tx.send(()).unwrap(); + resume_rx.await.unwrap(); + }) + .await + }); + snapshot_rx.await.unwrap(); + + let (clear_started_tx, clear_started_rx) = oneshot::channel(); + let clear_state = Arc::clone(&state); + let clear = tokio::spawn(async move { + mqtt_clear_messages_inner_with_hook(&clear_state, "mqtt-race".to_string(), || async move { + clear_started_tx.send(()).unwrap(); + }) + .await + }); + clear_started_rx.await.unwrap(); + assert!(!clear.is_finished()); + + resume_tx.send(()).unwrap(); + assert_eq!(poll.await.unwrap().unwrap().len(), 1); + clear.await.unwrap().unwrap(); + + assert!(client.get_messages(None, MQTT_LIVE_MESSAGE_LIMIT).await.is_empty()); + assert!(state.storage.load_mqtt_message_history("mqtt-race").await.unwrap().is_empty()); + } }