Self {
// Default log directory - will be overridden by frontend settings
@@ -170,7 +176,10 @@ impl SerialManager {
self.disconnect()?;
}
- let port = self.port_opener.open(port_name, &config)?;
+ // `mut` is only exercised by the POSIX write-timeout tweak below;
+ // Windows shares timeouts across cloned handles and leaves it alone.
+ #[allow(unused_mut)]
+ let mut port = self.port_opener.open(port_name, &config)?;
info!("Successfully opened serial port: {}", port_name);
// Reset and start reading thread
@@ -187,6 +196,15 @@ impl SerialManager {
let shutdown_flag = Arc::clone(&self.shutdown_flag);
let mut read_port = port.try_clone()?;
+ // Give the write side a longer timeout than the read-friendly 50 ms.
+ // On POSIX the timeout lives on each handle, so this does not slow
+ // the read loop; on Windows cloned handles share COMMTIMEOUTS, so we
+ // leave the write side at the builder's value there.
+ #[cfg(unix)]
+ if let Err(e) = port.set_timeout(Duration::from_millis(WRITE_TIMEOUT_MS)) {
+ warn!("Failed to set write timeout on {}: {}", port_name, e);
+ }
+
thread::spawn(move || {
let mut read_buffer = [0u8; 1024];
let initial_config = frame_segmentation_config.lock()
From 4cf035c7fe5e6507fa91bfbe834ff24f6aa4f233 Mon Sep 17 00:00:00 2001
From: gyanano <1624055384@qq.com>
Date: Wed, 2 Sep 2026 04:23:00 -0700
Subject: [PATCH 11/14] =?UTF-8?q?feat(serial):=20reader=20lifecycle=20?=
=?UTF-8?q?=E2=80=94=20bounded=20join,=20death=20detection,=20frame=20salv?=
=?UTF-8?q?age?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
RFC #3 Step 3:
- disconnect() returns the reader JoinHandle and the (bounded, 500ms) join
now happens OUTSIDE the big manager lock in the Tauri command, replacing
the in-lock 200ms sleep. connect() joins any stale reader defensively,
fixing the close->reopen EBUSY race.
- Reader-thread death is no longer silent: the error is recorded, the next
status poll (1s) flips is_connected=false with connection_error set, and
the frontend toasts 'Connection lost: ...' once per transition.
ConnectionStatus gains a connection_error field.
- Pending partial frames are flushed (salvaged) on ANY reader exit instead
of vanishing; new FrameSegmenter::flush().
- Fix scheduled-send zombie state: on disconnect or emptied payload the
SendPanel switch now resets instead of showing 'scheduled active' with a
dead timer.
Tests: 34/34 green (golden_reader_death_is_silent_today replaced by
reader_death_marks_connection_lost_and_salvages_pending; new
disconnect_returns_joinable_handle_and_reconnect_is_immediate +
flush_drains_pending_regardless_of_idle). Hardware-verified: unplug
detection + toast + switch reset, replug reconnect, rapid
disconnect/reconnect cycles, recording/export regression.
---
frontend/src/App.tsx | 9 +-
frontend/src/components/SendPanel.tsx | 5 +-
frontend/src/i18n/translations/en.json | 3 +-
frontend/src/i18n/translations/zh-CN.json | 3 +-
frontend/src/types.ts | 1 +
src-tauri/src/framing.rs | 24 ++++
src-tauri/src/main.rs | 16 ++-
src-tauri/src/serial_manager.rs | 135 +++++++++++++++++++---
src-tauri/src/types.rs | 4 +
9 files changed, 176 insertions(+), 24 deletions(-)
diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx
index 0730314..164f5d2 100644
--- a/frontend/src/App.tsx
+++ b/frontend/src/App.tsx
@@ -435,7 +435,14 @@ function App() {
const updateStatus = async () => {
try {
const status = await invoke
('get_connection_status');
- setConnectionStatus(status);
+ // Surface an unexpected loss (fatal read error, e.g. cable unplugged)
+ // exactly once per transition; manual disconnects carry no error.
+ setConnectionStatus((prev) => {
+ if (prev.is_connected && !status.is_connected && status.connection_error) {
+ toast.error(t('app.connectionLost').replace('{error}', status.connection_error));
+ }
+ return status;
+ });
} catch (error) {
console.error('Failed to get status:', error);
}
diff --git a/frontend/src/components/SendPanel.tsx b/frontend/src/components/SendPanel.tsx
index 841b65d..49d9b7d 100644
--- a/frontend/src/components/SendPanel.tsx
+++ b/frontend/src/components/SendPanel.tsx
@@ -226,10 +226,13 @@ const SendPanel: React.FC = ({
};
}, []);
- // Cleanup when disconnected or value becomes empty
+ // Cleanup when disconnected or value becomes empty: scheduled sending no
+ // longer has its preconditions, so reset the switch too — otherwise the
+ // panel shows "scheduled active" with a dead timer underneath.
useEffect(() => {
if (!isConnected || !value.trim()) {
stopScheduledSending();
+ setIsScheduledEnabled(false);
}
}, [isConnected, value]);
diff --git a/frontend/src/i18n/translations/en.json b/frontend/src/i18n/translations/en.json
index d52a16f..2bd99a5 100644
--- a/frontend/src/i18n/translations/en.json
+++ b/frontend/src/i18n/translations/en.json
@@ -1,7 +1,8 @@
{
"app": {
"title": "RSerial Debug Assistant",
- "subtitle": "Professional Tool"
+ "subtitle": "Professional Tool",
+ "connectionLost": "Connection lost: {error}"
},
"sidebar": {
"expandSidebar": "Expand Sidebar",
diff --git a/frontend/src/i18n/translations/zh-CN.json b/frontend/src/i18n/translations/zh-CN.json
index b6d8775..0984dc5 100644
--- a/frontend/src/i18n/translations/zh-CN.json
+++ b/frontend/src/i18n/translations/zh-CN.json
@@ -1,7 +1,8 @@
{
"app": {
"title": "RSerial Debug Assistant",
- "subtitle": "Professional Tool"
+ "subtitle": "Professional Tool",
+ "connectionLost": "连接意外断开：{error}"
},
"sidebar": {
"expandSidebar": "展开侧边栏",
diff --git a/frontend/src/types.ts b/frontend/src/types.ts
index 4a3dc90..7c5c635 100644
--- a/frontend/src/types.ts
+++ b/frontend/src/types.ts
@@ -54,6 +54,7 @@ export interface ConnectionStatus {
bytes_sent: number;
bytes_received: number;
connection_time: string | null;
+ connection_error?: string | null;
}
// Quick Command types
diff --git a/src-tauri/src/framing.rs b/src-tauri/src/framing.rs
index e0da76d..3bf16c0 100644
--- a/src-tauri/src/framing.rs
+++ b/src-tauri/src/framing.rs
@@ -115,9 +115,21 @@ impl FrameSegmenter {
}
/// Bytes currently pending (received but not yet framed).
+ #[cfg(test)]
pub fn pending(&self) -> &[u8] {
&self.buffer
}
+
+ /// Drain any buffered bytes as a final frame regardless of idle time.
+ /// Used when the reader is shutting down or dying so the tail of the
+ /// stream is not silently dropped (RFC #3 Step 3).
+ pub fn flush(&mut self) -> Option> {
+ if self.buffer.is_empty() {
+ None
+ } else {
+ Some(std::mem::take(&mut self.buffer))
+ }
+ }
}
/// Find the position of a delimiter in the data buffer.
@@ -339,4 +351,16 @@ mod tests {
assert_eq!(frames, vec![b"abcdefgh".to_vec(), b"\n".to_vec()]);
assert!(seg.pending().is_empty());
}
+
+ #[test]
+ fn flush_drains_pending_regardless_of_idle() {
+ let t0 = Instant::now();
+ let mut seg = FrameSegmenter::new(combined(FrameDelimiter::LF), t0);
+ assert!(seg.flush().is_none());
+ let frames = seg.feed(b"abc", t0);
+ assert!(frames.is_empty());
+ assert_eq!(seg.flush(), Some(b"abc".to_vec()));
+ assert!(seg.pending().is_empty());
+ assert!(seg.flush().is_none());
+ }
}
diff --git a/src-tauri/src/main.rs b/src-tauri/src/main.rs
index c05f867..33efda3 100644
--- a/src-tauri/src/main.rs
+++ b/src-tauri/src/main.rs
@@ -51,9 +51,17 @@ async fn connect_to_port(
#[tauri::command]
async fn disconnect_port(state: State<'_, AppState>) -> Result<(), String> {
- let mut manager = state.serial_manager.lock().unwrap();
- manager.disconnect()
- .map_err(|e| e.to_string())
+ // Fast state cleanup under the lock; the (bounded) reader join happens
+ // AFTER the lock is released so polling commands are never blocked
+ // behind a thread wait (RFC #3 Step 3).
+ let handle = {
+ let mut manager = state.serial_manager.lock().unwrap();
+ manager.disconnect().map_err(|e| e.to_string())?
+ };
+ if let Some(h) = handle {
+ SerialManager::join_reader_bounded(h);
+ }
+ Ok(())
}
#[tauri::command]
@@ -107,7 +115,7 @@ async fn send_data(
#[tauri::command]
async fn get_connection_status(state: State<'_, AppState>) -> Result {
- let manager = state.serial_manager.lock().unwrap();
+ let mut manager = state.serial_manager.lock().unwrap();
Ok(manager.get_status())
}
diff --git a/src-tauri/src/serial_manager.rs b/src-tauri/src/serial_manager.rs
index f2ce1af..ea4058f 100644
--- a/src-tauri/src/serial_manager.rs
+++ b/src-tauri/src/serial_manager.rs
@@ -34,6 +34,12 @@ pub struct SerialManager {
display_settings: Arc>,
// Port opening seam (system opener in production, scripted fake in tests)
port_opener: Arc,
+ // Reader thread lifecycle (RFC #3 Step 3)
+ reader_handle: Option>,
+ // Fatal read error written by the reader thread right before it dies
+ reader_error: Arc>>,
+ // Surfaced via ConnectionStatus until the next connect
+ connection_error: Option,
}
#[derive(Debug, Default)]
@@ -117,6 +123,9 @@ impl SerialManager {
timezone_offset_minutes: Arc::new(Mutex::new(0)),
display_settings: Arc::new(Mutex::new(DisplaySettings::default())),
port_opener: Arc::new(SystemPortOpener),
+ reader_handle: None,
+ reader_error: Arc::new(Mutex::new(None)),
+ connection_error: None,
}
}
@@ -173,8 +182,18 @@ impl SerialManager {
pub fn connect(&mut self, port_name: &str, config: SerialConfig) -> Result<()> {
if self.is_connected {
- self.disconnect()?;
+ let handle = self.disconnect()?;
+ if let Some(h) = handle {
+ Self::join_reader_bounded(h);
+ }
+ }
+ // Defensive: a reader that outlived a previous bounded join must be
+ // dead before the OS will let us reopen the same device (EBUSY race).
+ if let Some(h) = self.reader_handle.take() {
+ Self::join_reader_bounded(h);
}
+ *self.reader_error.lock().unwrap() = None;
+ self.connection_error = None;
// `mut` is only exercised by the POSIX write-timeout tweak below;
// Windows shares timeouts across cloned handles and leaves it alone.
@@ -194,6 +213,7 @@ impl SerialManager {
let display_settings = Arc::clone(&self.display_settings);
let port_name_clone = port_name.to_string();
let shutdown_flag = Arc::clone(&self.shutdown_flag);
+ let reader_error = Arc::clone(&self.reader_error);
let mut read_port = port.try_clone()?;
// Give the write side a longer timeout than the read-friendly 50 ms.
@@ -205,7 +225,7 @@ impl SerialManager {
warn!("Failed to set write timeout on {}: {}", port_name, e);
}
- thread::spawn(move || {
+ let reader_handle = thread::spawn(move || {
let mut read_buffer = [0u8; 1024];
let initial_config = frame_segmentation_config.lock()
.map(|guard| guard.clone())
@@ -305,12 +325,24 @@ impl SerialManager {
}
Err(e) => {
error!("Error reading from serial port: {}", e);
+ *reader_error.lock().unwrap() = Some(format!("{}", e));
break;
}
}
}
+
+ // Salvage the pending partial frame on ANY exit (shutdown flag or
+ // fatal error) so the tail of the stream is not silently dropped
+ // (RFC #3 Step 3).
+ if let Some(frame) = segmenter.flush() {
+ let disp_settings = display_settings.lock()
+ .map(|guard| guard.clone())
+ .unwrap_or_default();
+ emit_frame(frame, &disp_settings);
+ }
});
+ self.reader_handle = Some(reader_handle);
self.current_port = Some(port);
self.config = Some(config);
self.is_connected = true;
@@ -327,7 +359,10 @@ impl SerialManager {
Ok(())
}
- pub fn disconnect(&mut self) -> Result<()> {
+ /// Disconnect. Returns the reader thread handle so the CALLER can join
+ /// it outside the big manager lock (RFC #3 Step 3) — see
+ /// `join_reader_bounded`. State cleanup here is fast and non-blocking.
+ pub fn disconnect(&mut self) -> Result>> {
if self.is_connected {
// Signal reading thread to stop
self.shutdown_flag.store(true, Ordering::Relaxed);
@@ -335,14 +370,16 @@ impl SerialManager {
// Stop all recordings before disconnecting
self.stop_all_recordings();
- // Close the port first to force the reading thread to exit
+ // Close the write handle; the reader's cloned fd closes when the
+ // thread exits.
self.current_port = None;
- // Wait longer for thread to properly clean up
- thread::sleep(Duration::from_millis(200));
-
self.is_connected = false;
+ // A manual disconnect supersedes any concurrent reader death:
+ // don't surface it as an unexpected loss afterwards.
+ *self.reader_error.lock().unwrap() = None;
+
if let Some(port_name) = &self.port_name {
// Don't add disconnection log to reduce clutter
info!("Disconnected from {}", port_name);
@@ -358,7 +395,27 @@ impl SerialManager {
info!("Serial port disconnected");
}
- Ok(())
+ Ok(self.reader_handle.take())
+ }
+
+ /// Join a reader thread with a bounded wait. The reader wakes at least
+ /// every ~50 ms (port read timeout), so 500 ms is generous; if it still
+ /// has not exited we drop the handle (detaching the thread) rather than
+ /// block — the thread dies on its own once the shutdown flag is set and
+ /// the port fd is closed.
+ pub fn join_reader_bounded(handle: thread::JoinHandle<()>) {
+ let deadline = Instant::now() + Duration::from_millis(500);
+ loop {
+ if handle.is_finished() {
+ let _ = handle.join();
+ return;
+ }
+ if Instant::now() >= deadline {
+ warn!("Reader thread did not exit within 500ms; detaching");
+ return;
+ }
+ thread::sleep(Duration::from_millis(5));
+ }
}
pub fn send_data(&mut self, data: Vec) -> Result<()> {
@@ -405,13 +462,32 @@ impl SerialManager {
}
}
- pub fn get_status(&self) -> ConnectionStatus {
+ /// Lazy reader-death detection (RFC #3 Step 3): the reader thread records
+ /// a fatal error in `reader_error` before dying; the next status poll
+ /// (frontend: every second) turns that into a visible disconnect.
+ pub fn get_status(&mut self) -> ConnectionStatus {
+ if self.is_connected {
+ let death = self.reader_error.lock().unwrap().clone();
+ if let Some(err) = death {
+ warn!("Reader thread died, marking connection lost: {}", err);
+ self.connection_error = Some(err);
+ self.is_connected = false;
+ // Free the write handle and harvest the (finished) reader so
+ // the device can be reopened immediately.
+ self.current_port = None;
+ self.stop_all_recordings();
+ if let Some(h) = self.reader_handle.take() {
+ let _ = h.join();
+ }
+ }
+ }
+
let (bytes_sent, bytes_received, connection_time) = if let Ok(stats_guard) = self.stats.lock() {
(stats_guard.bytes_sent, stats_guard.bytes_received, stats_guard.connection_time)
} else {
(0, 0, None)
};
-
+
ConnectionStatus {
is_connected: self.is_connected,
port_name: self.port_name.clone(),
@@ -419,6 +495,7 @@ impl SerialManager {
bytes_sent,
bytes_received,
connection_time,
+ connection_error: self.connection_error.clone(),
}
}
@@ -1322,10 +1399,10 @@ mod tests {
}
#[test]
- fn golden_reader_death_is_silent_today() {
- // Pins the CURRENT (broken) behavior that RFC #3 Step 3 will fix:
- // a fatal read error kills the thread silently — is_connected stays
- // true and bytes pending in the accumulator are never framed.
+ fn reader_death_marks_connection_lost_and_salvages_pending() {
+ // RFC #3 Step 3: a fatal read error must surface — the next status
+ // poll reports disconnected with the error, and the pending partial
+ // frame is flushed instead of vanishing.
let port = ScriptedPort::new(
"SCRIPT",
vec![ScriptEvent::Bytes(b"abc".to_vec()), ScriptEvent::Fail],
@@ -1337,8 +1414,34 @@ mod tests {
// Give the thread time to hit Fail and break.
thread::sleep(Duration::from_millis(200));
- assert!(manager.get_status().is_connected); // the lie, pinned
- assert!(manager.get_logs().is_empty()); // pending "abc" lost, pinned
+
+ let status = manager.get_status();
+ assert!(!status.is_connected);
+ assert_eq!(status.connection_error.as_deref(), Some("scripted failure"));
+ // Pending "abc" salvaged as a final frame on death.
+ assert_eq!(frame_bytes(&manager.get_logs()), vec![b"abc".to_vec()]);
+
+ // A manual disconnect afterwards is a clean no-op.
+ assert!(manager.disconnect().unwrap().is_none());
+ assert!(manager.get_status().connection_error.is_some()); // sticky until next connect
+ }
+
+ #[test]
+ fn disconnect_returns_joinable_handle_and_reconnect_is_immediate() {
+ // RFC #3 Step 3: disconnect hands the reader handle to the caller;
+ // a bounded join then guarantees the device is free for reopen.
+ let port = ScriptedPort::new("SCRIPT", vec![]);
+ let mut manager = SerialManager::new()
+ .with_port_opener(Arc::new(ScriptedOpener { port }));
+ manager.connect("SCRIPT", SerialConfig::default()).unwrap();
+
+ let handle = manager.disconnect().unwrap().expect("reader handle");
+ SerialManager::join_reader_bounded(handle);
+
+ // Immediate reconnect must succeed (no stale reader, no EBUSY).
+ manager.connect("SCRIPT", SerialConfig::default()).unwrap();
+ assert!(manager.get_status().is_connected);
+ assert!(manager.get_status().connection_error.is_none());
manager.disconnect().unwrap();
}
diff --git a/src-tauri/src/types.rs b/src-tauri/src/types.rs
index 01b3e51..48ee5e3 100644
--- a/src-tauri/src/types.rs
+++ b/src-tauri/src/types.rs
@@ -109,6 +109,10 @@ pub struct ConnectionStatus {
pub bytes_sent: u64,
pub bytes_received: u64,
pub connection_time: Option>,
+ /// Fatal read error that ended the connection unexpectedly (RFC #3
+ /// Step 3). `None` for normal connects/disconnects; cleared on connect.
+ #[serde(default)]
+ pub connection_error: Option,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
From 8594c661e544d4528a6d976bb21ec2523e0ba992 Mon Sep 17 00:00:00 2001
From: gyanano <1624055384@qq.com>
Date: Wed, 2 Sep 2026 05:43:41 -0700
Subject: [PATCH 12/14] feat(serial): event push replaces 100ms log polling
(RFC #3 Step 4)
Backend:
- New bus.rs: Frame {session, seq, dir, t_mono_ns, t_wall, Arc data} with
per-session seq from 1 shared by TX/RX; FrameBus with bounded
per-subscriber queues (drop-oldest + dropped_before prefix-gap
accounting) and Nagle batching (GUI_DEFAULT 16ms/64KiB/512 frames/4096
queue; lone frame for an idle consumer pushes immediately past a 4ms
anti-storm interval).
- Read loop and send_data publish frames after the buffer write
(dual-write coexistence: log buffer/text+raw recording/stats unchanged,
all golden tests intact).
- New get_logs_snapshot command {epoch, session, entries}; clear_logs
returns a bumped epoch.
Bridge (main.rs): app-lifetime pump thread consumes batches with bounded
blocking recv, decorates (display_text/timestamp) and emits
serial://frames; the wire DTO carries no raw bytes (seq/dir/len +
display text).
Frontend:
- useSerialLogs hook: snapshot alignment, incremental append with rAF
throttling, seq dedupe, session-change resync, epoch guard so a snapshot
racing a clear never resurrects entries, dropped_before placeholder rows,
trim mirroring the backend 1000-entry cap; LogViewer keys rows by
session-seq.
- Rollback switch: localStorage serialEventPush=0 restores the legacy
100ms polling loop.
Tests: 40/40 green (5 new bus unit tests incl. drop-oldest prefix-gap
accounting; harness test asserting contiguous seq across RX/TX through
the real reader thread). Hardware-verified on FTDI loopback: burst
sending, clear-during-stream, search/autoscroll/counters, unplug.
---
frontend/src/App.tsx | 24 +-
frontend/src/components/LogViewer.tsx | 2 +-
frontend/src/hooks/useSerialLogs.ts | 172 +++++++++++
frontend/src/i18n/translations/en.json | 3 +-
frontend/src/i18n/translations/zh-CN.json | 3 +-
frontend/src/types.ts | 6 +
src-tauri/src/bus.rs | 345 ++++++++++++++++++++++
src-tauri/src/main.rs | 93 +++++-
src-tauri/src/serial_manager.rs | 101 ++++++-
src-tauri/src/types.rs | 17 ++
10 files changed, 748 insertions(+), 18 deletions(-)
create mode 100644 frontend/src/hooks/useSerialLogs.ts
create mode 100644 src-tauri/src/bus.rs
diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx
index 164f5d2..4424652 100644
--- a/frontend/src/App.tsx
+++ b/frontend/src/App.tsx
@@ -10,6 +10,7 @@ import SettingsModal from './components/SettingsModal';
import { SerialPortInfo, SerialConfig, LogEntry, ConnectionStatus, DataFormat, ChecksumConfig, QuickCommandList, QuickCommand, LineEnding, TextEncoding, FrameSegmentationConfig } from './types';
import { useTheme } from './contexts/ThemeContext';
import { useTranslation } from './i18n';
+import { useSerialLogs } from './hooks/useSerialLogs';
import { appendChecksum } from './utils/checksum';
import { loadTimezone, formatDateForFilename, getSystemTimezoneOffset, parseUtcOffset } from './utils/timezone';
import { Toaster } from './components/ui/sonner';
@@ -135,7 +136,12 @@ function App() {
bytes_received: 0,
connection_time: null,
});
- const [logs, setLogs] = useState([]);
+ const [polledLogs, setPolledLogs] = useState([]);
+ // RFC #3 Step 4: event push is the default log path; set
+ // localStorage 'serialEventPush' = '0' to roll back to 100ms polling.
+ const [eventPush] = useState(() => localStorage.getItem('serialEventPush') !== '0');
+ const { logs: pushedLogs, clearLogs: clearPushedLogs } = useSerialLogs(eventPush);
+ const logs = eventPush ? pushedLogs : polledLogs;
const [sendText, setSendText] = useState('');
const [sendFormat, setSendFormat] = useState('Text');
const [checksumConfig, setChecksumConfig] = useState({
@@ -417,15 +423,17 @@ function App() {
// Set up intervals for updating status, logs, and ports
const statusInterval = setInterval(updateStatus, 1000);
- const logsInterval = setInterval(updateLogs, 100); // More frequent log updates
+ // Event push (RFC #3 Step 4) replaces the 100ms full-clone log polling;
+ // the polling interval only runs in rollback mode.
+ const logsInterval = eventPush ? null : setInterval(updateLogs, 100);
const portsInterval = setInterval(loadPorts, 3000); // Check for new ports every 3 seconds
return () => {
clearInterval(statusInterval);
- clearInterval(logsInterval);
+ if (logsInterval !== null) clearInterval(logsInterval);
clearInterval(portsInterval);
};
- }, [loadPorts]); // 依赖loadPorts函数
+ }, [loadPorts, eventPush]); // 依赖loadPorts函数
const handlePortSelect = (port: string) => {
setSelectedPort(port);
@@ -451,7 +459,7 @@ function App() {
const updateLogs = async () => {
try {
const newLogs = await invoke('get_logs');
- setLogs(newLogs);
+ setPolledLogs(newLogs);
} catch (error) {
console.error('Failed to get logs:', error);
}
@@ -552,9 +560,13 @@ function App() {
};
const handleClearLogs = async () => {
+ if (eventPush) {
+ await clearPushedLogs();
+ return;
+ }
try {
await invoke('clear_logs');
- setLogs([]);
+ setPolledLogs([]);
} catch (error) {
console.error('Failed to clear logs:', error);
}
diff --git a/frontend/src/components/LogViewer.tsx b/frontend/src/components/LogViewer.tsx
index dd56c00..eec0626 100644
--- a/frontend/src/components/LogViewer.tsx
+++ b/frontend/src/components/LogViewer.tsx
@@ -664,7 +664,7 @@ const LogViewer: React.FC = ({ logs, onClear, onExport, isConnec
{logs.map((log, index) => (
{ logEntryRefs.current[index] = el; }}
className="py-1 px-2 rounded-[4px] transition-colors duration-150"
style={{
diff --git a/frontend/src/hooks/useSerialLogs.ts b/frontend/src/hooks/useSerialLogs.ts
new file mode 100644
index 0000000..4c35f71
--- /dev/null
+++ b/frontend/src/hooks/useSerialLogs.ts
@@ -0,0 +1,172 @@
+import { useState, useEffect, useRef, useCallback } from 'react';
+import { invoke } from '@tauri-apps/api/core';
+import { listen } from '@tauri-apps/api/event';
+import { LogEntry } from '../types';
+import { useTranslation } from '../i18n';
+
+/**
+ * Event-driven serial log state (RFC #3 Step 4): replaces the 100 ms
+ * full-clone polling of `get_logs` with a one-shot snapshot + incremental
+ * `serial://frames` batches.
+ *
+ * Invariants:
+ * - Snapshot carries (epoch, session); a snapshot that raced a `clear_logs`
+ * (epoch mismatch) or a newer snapshot (ticket) is discarded — cleared
+ * logs never resurrect.
+ * - Batches dedupe by seq: frames already covered by the snapshot or a
+ * prior batch are skipped; frames from before a clear are skipped.
+ * - A batch whose session differs from the current one triggers a resync
+ * snapshot instead of an append (seq restarts at 1 per session).
+ * - `dropped_before > 0` inserts a placeholder row so channel overload is
+ * visible instead of silent.
+ */
+
+interface FrameDto {
+ session: number;
+ seq: number;
+ direction: 'Sent' | 'Received';
+ len: number;
+ timestamp: string;
+ display_text: string;
+ timestamp_formatted: string | null;
+}
+
+interface FrameBatchDto {
+ session: number;
+ first_seq: number;
+ dropped_before: number;
+ frames: FrameDto[];
+}
+
+interface LogsSnapshot {
+ epoch: number;
+ session: number;
+ entries: LogEntry[];
+}
+
+/** Mirrors the backend default (`SerialManager::new`, clamped 100..10000). */
+const MAX_ENTRIES = 1000;
+
+export function useSerialLogs(enabled: boolean) {
+ const { t } = useTranslation();
+ const [logs, setLogs] = useState
([]);
+
+ const epochRef = useRef(0);
+ const sessionRef = useRef(0);
+ const lastSeqRef = useRef(0);
+ const clearedSeqRef = useRef(0);
+ const pendingRef = useRef([]);
+ const rafRef = useRef(null);
+ const snapTicketRef = useRef(0);
+
+ const flushPending = useCallback(() => {
+ rafRef.current = null;
+ if (pendingRef.current.length === 0) return;
+ const batch = pendingRef.current;
+ pendingRef.current = [];
+ setLogs((prev) => {
+ const next = [...prev, ...batch];
+ return next.length > MAX_ENTRIES ? next.slice(next.length - MAX_ENTRIES) : next;
+ });
+ }, []);
+
+ const scheduleFlush = useCallback(() => {
+ if (rafRef.current === null) {
+ rafRef.current = requestAnimationFrame(flushPending);
+ }
+ }, [flushPending]);
+
+ const takeSnapshot = useCallback(async () => {
+ const ticket = ++snapTicketRef.current;
+ try {
+ const snap = await invoke('get_logs_snapshot');
+ if (ticket !== snapTicketRef.current) return; // superseded by a newer snapshot
+ if (snap.epoch !== epochRef.current) return; // raced a clear: discard
+ epochRef.current = snap.epoch;
+ sessionRef.current = snap.session;
+ lastSeqRef.current = snap.entries.reduce((m, e) => Math.max(m, e.seq ?? 0), 0);
+ pendingRef.current = [];
+ setLogs(snap.entries);
+ } catch (error) {
+ console.error('Failed to take logs snapshot:', error);
+ }
+ }, []);
+
+ useEffect(() => {
+ if (!enabled) return;
+ let disposed = false;
+ let unlisten: (() => void) | undefined;
+
+ void takeSnapshot();
+
+ listen('serial://frames', (event) => {
+ if (disposed) return;
+ const b = event.payload;
+ if (b.session !== sessionRef.current) {
+ void takeSnapshot(); // session changed: seq restarted, resync instead of append
+ return;
+ }
+
+ const fresh: LogEntry[] = [];
+ if (b.dropped_before > 0) {
+ fresh.push({
+ timestamp: new Date().toISOString(),
+ direction: 'Received',
+ data: [],
+ format: 'Text',
+ port_name: '',
+ display_text: t('logViewer.framesDropped').replace('{n}', String(b.dropped_before)),
+ timestamp_formatted: undefined,
+ session: b.session,
+ gap_key: `gap-${b.session}-${b.first_seq}`,
+ });
+ }
+ for (const f of b.frames) {
+ // Skip frames already covered by the snapshot/prior batches, and
+ // frames that predate the last clear.
+ if (f.seq <= lastSeqRef.current || f.seq <= clearedSeqRef.current) continue;
+ fresh.push({
+ timestamp: f.timestamp,
+ direction: f.direction,
+ data: [],
+ format: 'Text',
+ port_name: '',
+ display_text: f.display_text,
+ timestamp_formatted: f.timestamp_formatted ?? undefined,
+ seq: f.seq,
+ session: f.session,
+ });
+ lastSeqRef.current = f.seq;
+ }
+ if (fresh.length > 0) {
+ pendingRef.current.push(...fresh);
+ scheduleFlush(); // rAF-throttled append
+ }
+ })
+ .then((u) => {
+ unlisten = u;
+ })
+ .catch((error) => console.error('Failed to listen serial://frames:', error));
+
+ return () => {
+ disposed = true;
+ unlisten?.();
+ if (rafRef.current !== null) cancelAnimationFrame(rafRef.current);
+ };
+ // eslint-disable-next-line react-hooks/exhaustive-deps
+ }, [enabled, takeSnapshot, scheduleFlush]);
+
+ const clearLogs = useCallback(async () => {
+ try {
+ const epoch = await invoke('clear_logs');
+ epochRef.current = epoch;
+ clearedSeqRef.current = lastSeqRef.current;
+ pendingRef.current = [];
+ setLogs([]);
+ } catch (error) {
+ console.error('Failed to clear logs:', error);
+ }
+ }, []);
+
+ return { logs, clearLogs };
+}
diff --git a/frontend/src/i18n/translations/en.json b/frontend/src/i18n/translations/en.json
index 2bd99a5..348ce3b 100644
--- a/frontend/src/i18n/translations/en.json
+++ b/frontend/src/i18n/translations/en.json
@@ -77,7 +77,8 @@
"tx": "TX",
"rx": "RX",
"total": "Total",
- "bytes": "Bytes"
+ "bytes": "Bytes",
+ "framesDropped": "⋯ dropped {n} frames (channel overloaded) ⋯"
},
"sendPanel": {
"payload": "Payload",
diff --git a/frontend/src/i18n/translations/zh-CN.json b/frontend/src/i18n/translations/zh-CN.json
index 0984dc5..830f237 100644
--- a/frontend/src/i18n/translations/zh-CN.json
+++ b/frontend/src/i18n/translations/zh-CN.json
@@ -77,7 +77,8 @@
"tx": "发送",
"rx": "接收",
"total": "总计",
- "bytes": "字节"
+ "bytes": "字节",
+ "framesDropped": "⋯ 丢失 {n} 帧（通道过载） ⋯"
},
"sendPanel": {
"payload": "发送数据",
diff --git a/frontend/src/types.ts b/frontend/src/types.ts
index 7c5c635..f55bce5 100644
--- a/frontend/src/types.ts
+++ b/frontend/src/types.ts
@@ -45,6 +45,12 @@ export interface LogEntry {
display_text: string;
/** Pre-formatted timestamp string (undefined if timestamps were disabled when entry was created) */
timestamp_formatted?: string;
+ /** Session-scoped sequence number (RFC #3 Step 4); 0/undefined for legacy entries */
+ seq?: number;
+ /** Owning session id */
+ session?: number;
+ /** Frontend-only marker for synthesized "frames dropped" placeholder rows */
+ gap_key?: string;
}
export interface ConnectionStatus {
diff --git a/src-tauri/src/bus.rs b/src-tauri/src/bus.rs
new file mode 100644
index 0000000..fece286
--- /dev/null
+++ b/src-tauri/src/bus.rs
@@ -0,0 +1,345 @@
+//! Frame bus: session/seq allocation, bounded per-subscriber queues with
+//! drop-oldest backpressure, and Nagle-style batching (RFC #3 Step 4).
+//!
+//! Contract pinned here:
+//! - `seq` is strictly increasing from 1 within a session; TX and RX share
+//! one sequence (transcript interleaving preserved).
+//! - A slow subscriber loses the OLDEST frames and every batch reports
+//! `dropped_before`, so gaps are always detectable (prefix gap + count).
+//! - The publisher (read thread / send path) never blocks: queue push is a
+//! µs-scale lock, never I/O.
+
+use crate::types::Direction;
+use chrono::{DateTime, Utc};
+use std::collections::VecDeque;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, Condvar, Mutex, OnceLock, Weak};
+use std::time::{Duration, Instant};
+
+pub type SessionId = u64;
+pub type Seq = u64;
+
+/// Process-level monotonic clock base (reserved for cross-port time alignment).
+fn mono_base() -> &'static Instant {
+ static BASE: OnceLock = OnceLock::new();
+ BASE.get_or_init(Instant::now)
+}
+
+fn mono_now_ns() -> u64 {
+ mono_base().elapsed().as_nanos() as u64
+}
+
+#[derive(Clone, Debug)]
+pub struct Frame {
+ pub session: SessionId,
+ pub seq: Seq,
+ pub dir: Direction,
+ /// Process-level monotonic timestamp, reserved for future cross-port
+ /// time alignment (RFC #3: must land now; adding it later would break
+ /// the transcript format).
+ #[allow(dead_code)]
+ pub t_mono_ns: u64,
+ pub t_wall: DateTime,
+ /// Arc-shared so fan-out to N subscribers is zero-copy.
+ pub data: Arc<[u8]>,
+}
+
+#[derive(Debug)]
+pub struct FrameBatch {
+ /// Session of the batch's LAST frame. A batch never mixes sessions in
+ /// practice (a session change requires a reconnect, which implies a
+ /// quiet port), but consumers should resync on session change anyway.
+ pub session: SessionId,
+ pub first_seq: Seq,
+ /// Frames dropped for THIS subscriber since the previous batch
+ /// (drop-oldest => the gap is always a prefix of the sequence).
+ pub dropped_before: u64,
+ pub frames: Vec,
+}
+
+#[derive(Debug, Clone, Copy)]
+pub struct BatchPolicy {
+ /// Max time the oldest pending frame may wait before flushing.
+ pub max_delay: Duration,
+ /// Minimum spacing between flushes; a lone frame for an idle consumer
+ /// pushes immediately only if the last flush is at least this old.
+ pub min_interval: Duration,
+ /// Flush when pending payload reaches this many bytes.
+ pub max_bytes: usize,
+ /// Flush when pending reaches this many frames.
+ pub max_frames: usize,
+ /// Bounded queue capacity per subscriber, in frames.
+ pub queue_frames: usize,
+}
+
+/// GUI default: ~60 fps worst case, IPC batches bounded by the frame cap.
+pub const GUI_DEFAULT: BatchPolicy = BatchPolicy {
+ max_delay: Duration::from_millis(16),
+ min_interval: Duration::from_millis(4),
+ max_bytes: 64 * 1024,
+ max_frames: 512,
+ queue_frames: 4096,
+};
+
+struct SubscriberState {
+ queue: VecDeque,
+ dropped: u64,
+}
+
+struct SubscriberShared {
+ state: Mutex,
+ cond: Condvar,
+ /// Bounded queue capacity in frames, from the subscriber's policy.
+ queue_frames: usize,
+}
+
+/// Receiving end of a subscription. The bridge contract is a pump thread
+/// calling `recv_batch` in a loop (bounded blocking, never async).
+pub struct Subscription {
+ shared: Arc,
+ policy: BatchPolicy,
+ last_flush: Mutex,
+}
+
+impl Subscription {
+ /// Block (bounded) until a batch is ready per the Nagle rules:
+ /// 1. idle consumer + lone frame + min_interval elapsed => push immediately
+ /// 2. otherwise accumulate until max_delay / max_bytes / max_frames
+ /// Returns `None` on timeout with no frames (lets the pump react to
+ /// external state); the subscription lives for the app's lifetime.
+ pub fn recv_batch(&self) -> Option {
+ let policy = &self.policy;
+ let mut st = self.shared.state.lock().unwrap();
+
+ // Wait for the first frame, bounded by max_delay.
+ while st.queue.is_empty() {
+ let (g, timed_out) = self
+ .shared
+ .cond
+ .wait_timeout(st, policy.max_delay)
+ .unwrap();
+ st = g;
+ if st.queue.is_empty() && timed_out.timed_out() {
+ return None;
+ }
+ }
+
+ let now = Instant::now();
+ let mut last_flush = self.last_flush.lock().unwrap();
+
+ // Rule 1: idle consumer, single frame, quiet period honored.
+ if st.queue.len() == 1 && now.duration_since(*last_flush) >= policy.min_interval {
+ let frame = st.queue.pop_front().unwrap();
+ let dropped_before = std::mem::take(&mut st.dropped);
+ *last_flush = Instant::now();
+ return Some(FrameBatch {
+ session: frame.session,
+ first_seq: frame.seq,
+ dropped_before,
+ frames: vec![frame],
+ });
+ }
+
+ // Rule 2: collect until a threshold or the max_delay deadline.
+ let mut frames: Vec = st.queue.drain(..).collect();
+ let mut bytes: usize = frames.iter().map(|f| f.data.len()).sum();
+ let deadline = Instant::now() + policy.max_delay;
+ loop {
+ if frames.len() >= policy.max_frames || bytes >= policy.max_bytes {
+ break;
+ }
+ let remaining = deadline.saturating_duration_since(Instant::now());
+ if remaining.is_zero() {
+ break;
+ }
+ let (g, _) = self.shared.cond.wait_timeout(st, remaining).unwrap();
+ st = g;
+ while let Some(f) = st.queue.pop_front() {
+ bytes += f.data.len();
+ frames.push(f);
+ }
+ if Instant::now() >= deadline {
+ break;
+ }
+ }
+
+ let dropped_before = std::mem::take(&mut st.dropped);
+ *last_flush = Instant::now();
+ let first = frames.first().expect("non-empty after wait");
+ Some(FrameBatch {
+ session: frames.last().unwrap().session,
+ first_seq: first.seq,
+ dropped_before,
+ frames: std::mem::take(&mut frames),
+ })
+ }
+}
+
+pub struct FrameBus {
+ session: AtomicU64,
+ seq: AtomicU64,
+ session_counter: AtomicU64,
+ subscribers: Mutex>>,
+}
+
+impl FrameBus {
+ pub fn new() -> Self {
+ Self {
+ session: AtomicU64::new(0),
+ seq: AtomicU64::new(0),
+ session_counter: AtomicU64::new(0),
+ subscribers: Mutex::new(Vec::new()),
+ }
+ }
+
+ /// Start a new session: id increments, seq resets so the next allocated
+ /// frame is seq 1. Called once per connect.
+ pub fn start_session(&self) -> SessionId {
+ let id = self.session_counter.fetch_add(1, Ordering::SeqCst) + 1;
+ self.seq.store(0, Ordering::SeqCst);
+ self.session.store(id, Ordering::SeqCst);
+ id
+ }
+
+ pub fn current_session(&self) -> SessionId {
+ self.session.load(Ordering::SeqCst)
+ }
+
+ /// Allocate a frame (assigns session/seq/timestamps, wraps data in Arc).
+ /// Separate from `publish` so the caller can also record the same
+ /// seq/session into its own structures (e.g. LogEntry) before fan-out.
+ pub fn alloc_frame(&self, dir: Direction, data: Vec) -> Frame {
+ Frame {
+ session: self.session.load(Ordering::SeqCst),
+ seq: self.seq.fetch_add(1, Ordering::SeqCst) + 1,
+ dir,
+ t_mono_ns: mono_now_ns(),
+ t_wall: Utc::now(),
+ data: Arc::from(data.into_boxed_slice()),
+ }
+ }
+
+ /// Fan out to all live subscribers. Never blocks on I/O; a full queue
+ /// drops the OLDEST frame and counts it for `dropped_before`.
+ pub fn publish(&self, frame: &Frame) {
+ let mut subs = self.subscribers.lock().unwrap();
+ subs.retain(|weak| {
+ if let Some(shared) = weak.upgrade() {
+ let mut st = shared.state.lock().unwrap();
+ if st.queue.len() >= shared.queue_frames {
+ st.queue.pop_front();
+ st.dropped += 1;
+ }
+ st.queue.push_back(frame.clone());
+ drop(st);
+ shared.cond.notify_one();
+ true
+ } else {
+ false // prune dead subscriptions
+ }
+ });
+ }
+
+ pub fn subscribe(&self, policy: BatchPolicy) -> Subscription {
+ let shared = Arc::new(SubscriberShared {
+ state: Mutex::new(SubscriberState {
+ queue: VecDeque::new(),
+ dropped: 0,
+ }),
+ cond: Condvar::new(),
+ queue_frames: policy.queue_frames,
+ });
+ self.subscribers.lock().unwrap().push(Arc::downgrade(&shared));
+ Subscription {
+ shared,
+ policy,
+ // Allow the very first frame to push immediately.
+ last_flush: Mutex::new(Instant::now() - policy.min_interval),
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn fast_policy() -> BatchPolicy {
+ BatchPolicy {
+ max_delay: Duration::from_millis(5),
+ min_interval: Duration::from_millis(2),
+ max_bytes: 1024,
+ max_frames: 3,
+ queue_frames: 4,
+ }
+ }
+
+ #[test]
+ fn seq_is_strictly_monotonic_and_resets_per_session() {
+ let bus = FrameBus::new();
+ let s1 = bus.start_session();
+ let f1 = bus.alloc_frame(Direction::Received, b"a".to_vec());
+ let f2 = bus.alloc_frame(Direction::Sent, b"b".to_vec());
+ assert_eq!((f1.session, f1.seq), (s1, 1));
+ assert_eq!((f2.session, f2.seq), (s1, 2)); // TX/RX share one sequence
+
+ let s2 = bus.start_session();
+ assert!(s2 > s1);
+ let f3 = bus.alloc_frame(Direction::Received, b"c".to_vec());
+ assert_eq!((f3.session, f3.seq), (s2, 1));
+ }
+
+ #[test]
+ fn lone_frame_pushes_immediately_for_idle_consumer() {
+ let bus = FrameBus::new();
+ bus.start_session();
+ let sub = bus.subscribe(fast_policy());
+ bus.publish(&bus.alloc_frame(Direction::Received, b"hi".to_vec()));
+ let start = Instant::now();
+ let batch = sub.recv_batch().unwrap();
+ assert!(start.elapsed() < Duration::from_millis(5));
+ assert_eq!(batch.first_seq, 1);
+ assert_eq!(batch.frames.len(), 1);
+ assert_eq!(batch.dropped_before, 0);
+ }
+
+ #[test]
+ fn queued_frames_collect_into_one_batch_without_waiting() {
+ let bus = FrameBus::new();
+ bus.start_session();
+ let sub = bus.subscribe(fast_policy());
+ for i in 0..3u8 {
+ bus.publish(&bus.alloc_frame(Direction::Received, vec![i]));
+ }
+ // 3 frames queued > lone-frame case: drained as one batch, no delay.
+ let batch = sub.recv_batch().unwrap();
+ assert_eq!(batch.frames.len(), 3);
+ assert_eq!(batch.first_seq, 1);
+ }
+
+ #[test]
+ fn slow_subscriber_drops_oldest_and_counts_prefix_gap() {
+ let bus = FrameBus::new();
+ bus.start_session();
+ let sub = bus.subscribe(fast_policy()); // queue_frames = 4
+ // Publish 6 without draining: frames 1-2 are dropped as oldest.
+ for i in 0..6u32 {
+ bus.publish(&bus.alloc_frame(Direction::Received, i.to_le_bytes().to_vec()));
+ }
+ let batch = sub.recv_batch().unwrap();
+ assert_eq!(batch.dropped_before, 2);
+ assert_eq!(batch.first_seq, 3);
+ let seqs: Vec = batch.frames.iter().map(|f| f.seq).collect();
+ assert_eq!(seqs, vec![3, 4, 5, 6]);
+ }
+
+ #[test]
+ fn recv_batch_times_out_when_quiet() {
+ let bus = FrameBus::new();
+ bus.start_session();
+ let sub = bus.subscribe(fast_policy());
+ let start = Instant::now();
+ assert!(sub.recv_batch().is_none());
+ let elapsed = start.elapsed();
+ assert!(elapsed >= Duration::from_millis(5) && elapsed < Duration::from_millis(50));
+ }
+}
diff --git a/src-tauri/src/main.rs b/src-tauri/src/main.rs
index 33efda3..5125deb 100644
--- a/src-tauri/src/main.rs
+++ b/src-tauri/src/main.rs
@@ -6,6 +6,7 @@ use std::sync::Mutex;
use tauri::State;
mod serial_manager;
+mod bus;
mod framing;
mod types;
mod updater;
@@ -14,6 +15,79 @@ use serial_manager::SerialManager;
use types::*;
use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
+// ── RFC #3 Step 4: event push bridge ─────────────────────────────────
+/// Wire DTO for `serial://frames`: the hot path carries no raw bytes
+/// (seq/dir/len + decorated display text); raw bytes remain available via
+/// the snapshot command and export.
+#[derive(Clone, serde::Serialize)]
+struct FrameDto {
+ session: u64,
+ seq: u64,
+ direction: Direction,
+ len: usize,
+ timestamp: chrono::DateTime,
+ display_text: String,
+ timestamp_formatted: Option,
+}
+
+#[derive(Clone, serde::Serialize)]
+struct FrameBatchDto {
+ session: u64,
+ first_seq: u64,
+ dropped_before: u64,
+ frames: Vec,
+}
+
+/// Bridge contract: a plain pump thread with bounded blocking recv; the
+/// subscription lives for the app's lifetime (sessions come and go).
+fn spawn_frame_pump(app: &tauri::App) {
+ use tauri::{Emitter, Manager};
+ let app_handle = app.handle().clone();
+ let state = app.state::();
+ let (bus, disp, tz) = {
+ let manager = state.serial_manager.lock().unwrap();
+ (
+ manager.bus(),
+ manager.display_settings_handle(),
+ manager.timezone_offset_handle(),
+ )
+ };
+ std::thread::spawn(move || {
+ let sub = bus.subscribe(bus::GUI_DEFAULT);
+ loop {
+ let Some(batch) = sub.recv_batch() else {
+ continue; // quiet timeout tick
+ };
+ let settings = disp.lock().map(|g| g.clone()).unwrap_or_default();
+ let tz_offset = *tz.lock().unwrap_or_else(|e| e.into_inner());
+ let frames: Vec = batch
+ .frames
+ .iter()
+ .map(|f| FrameDto {
+ session: f.session,
+ seq: f.seq,
+ direction: f.dir,
+ len: f.data.len(),
+ timestamp: f.t_wall,
+ display_text: serial_manager::format_data_for_display(&f.data, &settings),
+ timestamp_formatted: if settings.show_timestamps {
+ Some(serial_manager::format_timestamp_with_offset(tz_offset))
+ } else {
+ None
+ },
+ })
+ .collect();
+ let dto = FrameBatchDto {
+ session: batch.session,
+ first_seq: batch.first_seq,
+ dropped_before: batch.dropped_before,
+ frames,
+ };
+ let _ = app_handle.emit("serial://frames", dto);
+ }
+ });
+}
+
// Application state
struct AppState {
serial_manager: Mutex,
@@ -125,11 +199,19 @@ async fn get_logs(state: State<'_, AppState>) -> Result, String> {
Ok(manager.get_logs())
}
+/// Initial alignment for the event-driven log view (RFC #3 Step 4).
#[tauri::command]
-async fn clear_logs(state: State<'_, AppState>) -> Result<(), String> {
+async fn get_logs_snapshot(state: State<'_, AppState>) -> Result {
+ let manager = state.serial_manager.lock().unwrap();
+ Ok(manager.get_logs_snapshot())
+}
+
+/// Returns the new log epoch; the frontend uses it to discard snapshots
+/// that raced this clear.
+#[tauri::command]
+async fn clear_logs(state: State<'_, AppState>) -> Result {
let mut manager = state.serial_manager.lock().unwrap();
- manager.clear_logs();
- Ok(())
+ Ok(manager.clear_logs())
}
#[tauri::command]
@@ -360,6 +442,10 @@ fn main() {
tauri::Builder::default()
.manage(AppState::default())
.plugin(tauri_plugin_dialog::init())
+ .setup(|app| {
+ spawn_frame_pump(app);
+ Ok(())
+ })
.invoke_handler(tauri::generate_handler![
list_serial_ports,
connect_to_port,
@@ -367,6 +453,7 @@ fn main() {
send_data,
get_connection_status,
get_logs,
+ get_logs_snapshot,
clear_logs,
export_logs,
save_session,
diff --git a/src-tauri/src/serial_manager.rs b/src-tauri/src/serial_manager.rs
index ea4058f..2dbbd1e 100644
--- a/src-tauri/src/serial_manager.rs
+++ b/src-tauri/src/serial_manager.rs
@@ -40,6 +40,12 @@ pub struct SerialManager {
reader_error: Arc>>,
// Surfaced via ConnectionStatus until the next connect
connection_error: Option,
+ // Frame bus for event push (RFC #3 Step 4); read/send paths publish,
+ // the bridge pump thread consumes and emits `serial://frames`.
+ bus: Arc,
+ // Bumped on every clear_logs; snapshots carry it to detect
+ // clear-during-snapshot resurrection.
+ log_epoch: u64,
}
#[derive(Debug, Default)]
@@ -126,6 +132,8 @@ impl SerialManager {
reader_handle: None,
reader_error: Arc::new(Mutex::new(None)),
connection_error: None,
+ bus: Arc::new(crate::bus::FrameBus::new()),
+ log_epoch: 0,
}
}
@@ -201,6 +209,9 @@ impl SerialManager {
let mut port = self.port_opener.open(port_name, &config)?;
info!("Successfully opened serial port: {}", port_name);
+ // New session: seq restarts at 1 (RFC #3 Step 4).
+ self.bus.start_session();
+
// Reset and start reading thread
self.shutdown_flag.store(false, Ordering::Relaxed);
let logs = Arc::clone(&self.logs);
@@ -214,6 +225,7 @@ impl SerialManager {
let port_name_clone = port_name.to_string();
let shutdown_flag = Arc::clone(&self.shutdown_flag);
let reader_error = Arc::clone(&self.reader_error);
+ let bus = Arc::clone(&self.bus);
let mut read_port = port.try_clone()?;
// Give the write side a longer timeout than the read-friendly 50 ms.
@@ -233,7 +245,8 @@ impl SerialManager {
let mut segmenter = FrameSegmenter::new(initial_config, Instant::now());
// Single frame-emission path (replaces the four duplicated
- // blocks): text recording -> display formatting -> log buffer -> stats.
+ // blocks): text recording -> display formatting -> log buffer
+ // -> stats -> bus publish (dual-write 存续期, RFC #3 Step 4).
let emit_frame = |frame_data: Vec, disp_settings: &DisplaySettings| {
// Write to text recording file with timestamp and RX label
if let Ok(mut guard) = text_file.lock() {
@@ -245,6 +258,10 @@ impl SerialManager {
}
}
+ // Allocate the bus frame first so the LogEntry carries the
+ // same seq/session the event subscribers see.
+ let frame = bus.alloc_frame(Direction::Received, frame_data.clone());
+
// Format display text and timestamp based on current settings
let tz_offset = *timezone_offset.lock().unwrap_or_else(|e| e.into_inner());
let display_text = format_data_for_display(&frame_data, disp_settings);
@@ -263,6 +280,8 @@ impl SerialManager {
port_name: port_name_clone.clone(),
display_text,
timestamp_formatted,
+ seq: frame.seq,
+ session: frame.session,
};
if let Ok(mut logs_guard) = logs.lock() {
@@ -276,6 +295,11 @@ impl SerialManager {
if let Ok(mut stats_guard) = stats.lock() {
stats_guard.bytes_received += data_len;
}
+
+ // Event fan-out comes last: the buffer write must land first
+ // so a snapshot racing this batch never misses the entry
+ // (subscribers dedupe by seq anyway).
+ bus.publish(&frame);
};
loop {
@@ -435,6 +459,10 @@ impl SerialManager {
stats_guard.bytes_sent += data.len() as u64;
}
+ // TX shares the session's single seq sequence (transcript
+ // interleaving preserved, RFC #3 Step 4).
+ let frame = self.bus.alloc_frame(Direction::Sent, data.clone());
+
// Get current display settings for formatting
let disp_settings = self.get_display_settings();
let tz_offset = *self.timezone_offset_minutes.lock().unwrap_or_else(|e| e.into_inner());
@@ -454,8 +482,13 @@ impl SerialManager {
port_name: self.port_name.clone().unwrap_or_default(),
display_text,
timestamp_formatted,
+ seq: frame.seq,
+ session: frame.session,
});
+ // Buffer-first, then fan out (see emit_frame).
+ self.bus.publish(&frame);
+
Ok(())
} else {
Err(anyhow!("No port available"))
@@ -507,10 +540,38 @@ impl SerialManager {
}
}
- pub fn clear_logs(&mut self) {
+ pub fn clear_logs(&mut self) -> u64 {
if let Ok(mut logs) = self.logs.lock() {
logs.clear();
}
+ // Bump the epoch so any snapshot started before this clear is
+ // recognized as stale by the frontend (no resurrection).
+ self.log_epoch += 1;
+ self.log_epoch
+ }
+
+ /// Initial-alignment snapshot for the event-driven log view.
+ pub fn get_logs_snapshot(&self) -> LogsSnapshot {
+ LogsSnapshot {
+ epoch: self.log_epoch,
+ session: self.bus.current_session(),
+ entries: self.get_logs(),
+ }
+ }
+
+ /// Event bus handle for the bridge pump (RFC #3 Step 4).
+ pub fn bus(&self) -> Arc {
+ Arc::clone(&self.bus)
+ }
+
+ /// Shared display settings, for the bridge pump's decoration pass.
+ pub fn display_settings_handle(&self) -> Arc> {
+ Arc::clone(&self.display_settings)
+ }
+
+ /// Shared timezone offset, for the bridge pump's decoration pass.
+ pub fn timezone_offset_handle(&self) -> Arc> {
+ Arc::clone(&self.timezone_offset_minutes)
}
pub fn export_logs(&self, file_path: &str, format: ExportFormat, timezone_offset_minutes: i32) -> Result<()> {
@@ -857,7 +918,7 @@ impl SerialManager {
}
/// Format current UTC time with timezone offset applied
-fn format_timestamp_with_offset(offset_minutes: i32) -> String {
+pub(crate) fn format_timestamp_with_offset(offset_minutes: i32) -> String {
use chrono::FixedOffset;
let offset_seconds = offset_minutes * 60;
let tz_offset = FixedOffset::east_opt(offset_seconds).unwrap_or_else(|| FixedOffset::east_opt(0).unwrap());
@@ -1021,7 +1082,7 @@ fn sort_usb_ports_first(mut ports: Vec) -> Vec {
}
/// Format data based on display settings
-fn format_data_for_display(data: &[u8], settings: &DisplaySettings) -> String {
+pub(crate) fn format_data_for_display(data: &[u8], settings: &DisplaySettings) -> String {
match settings.format {
ReceiveDisplayFormat::Hex => format_bytes_as_hex(data),
ReceiveDisplayFormat::Txt => format_bytes_as_text(data, &settings.encoding, &settings.special_char_config),
@@ -1427,8 +1488,36 @@ mod tests {
}
#[test]
- fn disconnect_returns_joinable_handle_and_reconnect_is_immediate() {
- // RFC #3 Step 3: disconnect hands the reader handle to the caller;
+ fn bus_events_carry_contiguous_seq_across_rx_and_tx() {
+ // RFC #3 Step 4 acceptance: frames reach subscribers with contiguous
+ // seqs, TX and RX sharing one sequence, no loss, no duplication.
+ let port = ScriptedPort::new("SCRIPT", vec![ScriptEvent::Bytes(b"hello".to_vec())]);
+ let mut manager = SerialManager::new()
+ .with_port_opener(Arc::new(ScriptedOpener { port }));
+ manager.set_frame_segmentation_config(seg_timeout());
+ let sub = manager.bus().subscribe(crate::bus::GUI_DEFAULT);
+ manager.connect("SCRIPT", SerialConfig::default()).unwrap();
+
+ thread::sleep(Duration::from_millis(150)); // RX arrives + idle flush
+ manager.send_data(b"ping".to_vec()).unwrap();
+
+ let mut seen: Vec<(u64, Direction)> = Vec::new();
+ let deadline = Instant::now() + Duration::from_secs(2);
+ while Instant::now() < deadline && seen.len() < 2 {
+ if let Some(batch) = sub.recv_batch() {
+ seen.extend(batch.frames.iter().map(|f| (f.seq, f.dir)));
+ }
+ }
+ manager.disconnect().unwrap();
+
+ assert_eq!(
+ seen,
+ vec![(1, Direction::Received), (2, Direction::Sent)]
+ );
+ }
+
+ #[test]
+ fn disconnect_returns_joinable_handle_and_reconnect_is_immediate() { // RFC #3 Step 3: disconnect hands the reader handle to the caller;
// a bounded join then guarantees the device is free for reopen.
let port = ScriptedPort::new("SCRIPT", vec![]);
let mut manager = SerialManager::new()
diff --git a/src-tauri/src/types.rs b/src-tauri/src/types.rs
index 48ee5e3..ba20b88 100644
--- a/src-tauri/src/types.rs
+++ b/src-tauri/src/types.rs
@@ -93,6 +93,23 @@ pub struct LogEntry {
pub display_text: String,
/// Pre-formatted timestamp string (None if timestamps were disabled when entry was created)
pub timestamp_formatted: Option,
+ /// Session-scoped sequence number, strictly increasing from 1 per
+ /// session, TX and RX sharing one sequence (RFC #3 Step 4).
+ #[serde(default)]
+ pub seq: u64,
+ /// Session this entry belongs to (0 = pre-event-model legacy entries).
+ #[serde(default)]
+ pub session: u64,
+}
+
+/// Initial-alignment snapshot for the event-driven log view (RFC #3 Step 4).
+/// `epoch` guards against clear-during-snapshot resurrection: a snapshot
+/// taken before a `clear_logs` carries a stale epoch and must be discarded.
+#[derive(Debug, Clone, Serialize)]
+pub struct LogsSnapshot {
+ pub epoch: u64,
+ pub session: u64,
+ pub entries: Vec,
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
From a8786fff10ac4fc41748ab7d7f32523bab5f7e60 Mon Sep 17 00:00:00 2001
From: gyanano <1624055384@qq.com>
Date: Wed, 2 Sep 2026 05:46:19 -0700
Subject: [PATCH 13/14] feat(ui): scheduled-send interval down to 10ms +
draft-based editing
- Lower the interval floor from 100ms to 10ms (the event-push RX path
keeps up now; the rate guard still warns when the line can't).
- Fix mid-edit clamping: the input keeps a draft string while focused and
only commits (clamped) on blur/Enter; Escape cancels. Typing '50' no
longer snaps to the minimum after the first digit.
---
frontend/src/components/SendPanel.tsx | 39 +++++++++++++++++++++------
1 file changed, 31 insertions(+), 8 deletions(-)
diff --git a/frontend/src/components/SendPanel.tsx b/frontend/src/components/SendPanel.tsx
index 49d9b7d..4c2584f 100644
--- a/frontend/src/components/SendPanel.tsx
+++ b/frontend/src/components/SendPanel.tsx
@@ -89,6 +89,10 @@ const SendPanel: React.FC = ({
const [sendMode, setSendMode] = useState('normal');
const [isScheduledEnabled, setIsScheduledEnabled] = useState(false);
const [scheduledInterval, setScheduledInterval] = useState(1000);
+ // Draft string while the interval input is being edited: typing is never
+ // clamped mid-edit; the value commits (clamped) on blur/Enter, Escape
+ // cancels. Prevents "5" snapping to the minimum while typing "50".
+ const [intervalDraft, setIntervalDraft] = useState(null);
const [isScheduledRunning, setIsScheduledRunning] = useState(false);
const intervalRef = useRef | null>(null);
const [isChecksumExpanded, setIsChecksumExpanded] = useState(false);
@@ -189,9 +193,20 @@ const SendPanel: React.FC = ({
}
};
- const handleIntervalChange = (e: React.ChangeEvent) => {
- const newInterval = Math.max(100, parseInt(e.target.value) || 1000);
- setScheduledInterval(newInterval);
+ // Interval bounds: 10ms floor (event-push RX path can keep up now),
+ // 60s ceiling, 1s fallback for empty/invalid drafts.
+ const MIN_INTERVAL_MS = 10;
+ const MAX_INTERVAL_MS = 60000;
+
+ const commitIntervalDraft = () => {
+ if (intervalDraft === null) return;
+ const parsed = parseInt(intervalDraft, 10);
+ const next = Number.isNaN(parsed)
+ ? scheduledInterval
+ : Math.min(MAX_INTERVAL_MS, Math.max(MIN_INTERVAL_MS, parsed));
+ setIntervalDraft(null);
+ if (next === scheduledInterval) return;
+ setScheduledInterval(next);
// Restart interval with new timing if already running
if (isScheduledRunning) {
@@ -200,6 +215,8 @@ const SendPanel: React.FC = ({
}
};
+ const cancelIntervalDraft = () => setIntervalDraft(null);
+
const startScheduledSending = () => {
if (!value.trim() || !isConnected) return;
@@ -587,12 +604,18 @@ const SendPanel: React.FC = ({
setIntervalDraft(String(scheduledInterval))}
+ onChange={(e) => setIntervalDraft(e.target.value.replace(/[^0-9]/g, ''))}
+ onBlur={commitIntervalDraft}
+ onKeyDown={(e) => {
+ if (e.key === 'Enter') commitIntervalDraft();
+ if (e.key === 'Escape') cancelIntervalDraft();
+ }}
className="w-full h-7 text-xs text-right"
- min={100}
- max={60000}
- step={100}
+ min={MIN_INTERVAL_MS}
+ max={MAX_INTERVAL_MS}
+ step={10}
disabled={!isConnected || isScheduledRunning}
/>
From af8cccc9ad5f6ce7c5d90909250457ef5f7fdd8e Mon Sep 17 00:00:00 2001
From: gyanano <1624055384@qq.com>
Date: Wed, 2 Sep 2026 05:54:15 -0700
Subject: [PATCH 14/14] chore(release): v1.4.0
---
frontend/package.json | 2 +-
src-tauri/Cargo.toml | 2 +-
src-tauri/tauri.conf.json | 2 +-
version.json | 4 ++--
4 files changed, 5 insertions(+), 5 deletions(-)
diff --git a/frontend/package.json b/frontend/package.json
index 37f9471..fe463a5 100644
--- a/frontend/package.json
+++ b/frontend/package.json
@@ -1,7 +1,7 @@
{
"type": "module",
"name": "serial-debug-assistant-frontend",
- "version": "1.3.3",
+ "version": "1.4.0",
"description": "Beautiful frontend for RSerial Debug Assistant",
"scripts": {
"dev": "vite",
diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml
index 7fb9f04..2ea8224 100644
--- a/src-tauri/Cargo.toml
+++ b/src-tauri/Cargo.toml
@@ -1,6 +1,6 @@
[package]
name = "serial-debug-assistant"
-version = "1.3.3"
+version = "1.4.0"
description = "Cross-platform serial debugging assistant"
authors = ["Gyanano"]
license = "MIT"
diff --git a/src-tauri/tauri.conf.json b/src-tauri/tauri.conf.json
index 9d33a3a..8f9de1d 100644
--- a/src-tauri/tauri.conf.json
+++ b/src-tauri/tauri.conf.json
@@ -1,7 +1,7 @@
{
"$schema": "https://schema.tauri.app/config/2.0.0",
"productName": "RSerial Debug Assistant",
- "version": "1.3.3",
+ "version": "1.4.0",
"identifier": "Gyanano",
"build": {
"frontendDist": "../frontend/dist",
diff --git a/version.json b/version.json
index ba242fc..c9fae75 100644
--- a/version.json
+++ b/version.json
@@ -1,4 +1,4 @@
{
- "version": "1.3.3",
- "build": "20260819"
+ "version": "1.4.0",
+ "build": "20260902"
}