fix(mqtt): preserve console history across disconnects

2026-10-21 04:47:00 +00:00 · 2026-10-12 19:50:43 +08:00
parent 4a4a7603b5
commit 177b9a3b39
7 changed files with 587 additions and 14 deletions
@@ -64,6 +64,7 @@ const formNoLocal = ref(restoredUiState.formNoLocal ?? false);
 const formEnabled = ref(restoredUiState.formEnabled ?? true);
 const editingTopic = ref<string | null>(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<MqttBrokerInfo>,
      mqttListTopics(props.connectionId) as Promise<[string, string][]>,
      mqttListSavedTopicConfigs(props.connectionId) as Promise<MqttSavedTopic[]>,
-      mqttGetMessages(props.connectionId, selectedTopic.value || undefined, 50) as Promise<MqttMessage[]>,
+      mqttGetMessages(props.connectionId, selectedTopic.value || undefined, MQTT_MESSAGE_HISTORY_LIMIT) as Promise<MqttMessage[]>,
    ]);
    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 */
@@ -28,11 +28,13 @@ const publishPanelRef = ref<HTMLElement | null>(null);
 const payloadTextareaRef = ref<HTMLTextAreaElement | null>(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;
 }
 </script>

@@ -326,6 +326,8 @@ pub struct MqttClient {
    publish_send_lock: Mutex<()>,
    /// 最近接收的消息缓冲区（保留最近 N 条）
    message_buffer: RwLock<Vec<MqttMessage>>,
+    /// 串行化消息缓冲区与持久化历史之间的复合操作。
+    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<MqttMessage, String> {
        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<MqttMessage>) -> Arc<Self> {
+        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;
@@ -41,7 +41,7 @@ pub async fn unsubscribe(client: &Arc<MqttClient>, topic: &str) -> Result<(), St
 }

 /// 发布消息
-pub async fn publish(client: &Arc<MqttClient>, request: &MqttPublishRequest) -> Result<(), String> {
+pub async fn publish(client: &Arc<MqttClient>, request: &MqttPublishRequest) -> Result<MqttMessage, String> {
    client.publish(request).await
 }

@@ -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;
@@ -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<String>, 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<Vec<u8>> = 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<MqttMessage> = 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<String>, 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<Vec<MqttMessage>, 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<MqttMessage> = 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");
@@ -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<String>, 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::<Vec<_>>();
+    let filter_segments = filter.split('/').collect::<Vec<_>>();
+    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<String>,
    limit: Option<usize>,
 ) -> Result<Vec<MqttMessage>, 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<String>,
+    limit: Option<usize>,
+) -> Result<Vec<MqttMessage>, String> {
+    mqtt_get_messages_inner_with_hook(state, connection_id, topic_filter, limit, || async {}).await
+}
+
+async fn mqtt_get_messages_inner_with_hook<F, Fut>(
+    state: &AppState,
+    connection_id: String,
+    topic_filter: Option<String>,
+    limit: Option<usize>,
+    after_snapshot: F,
+) -> Result<Vec<MqttMessage>, String>
+where
+    F: FnOnce() -> Fut,
+    Fut: Future<Output = ()>,
+{
+    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::<MqttMessageIdentity, usize>::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<AppState>>, 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<F, Fut>(
+    state: &AppState,
+    connection_id: String,
+    before_lock: F,
+) -> Result<(), String>
+where
+    F: FnOnce() -> Fut,
+    Fut: Future<Output = ()>,
+{
+    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());
+    }
 }