mirror of
https://github.com/spartanz51/tutabridge.git
synced 2026-06-24 10:54:32 +02:00
Drop the last two timers: prefetch and the UI stats poll go event-driven
Phase 2.5 — `prefetch_loop` no longer wakes every 30s. It subscribes to `MailStore::subscribe()` and only catches up missing bodies when the generation counter ticks (event-bus delta, sync_folder finishing, bootstrap). When the store is quiet — every cached mail has its .eml on disk — the loop sleeps indefinitely on the watch channel. The `PREFETCH_INTERVAL` constant is gone. Phase 2.6 — the dashboard 1s `setInterval` is gone too. `BridgeHandle` gets a `stats_dirty_tx: broadcast::Sender<()>` that pulses on every `MailStore` bump, every `WsState` transition, and the start / stop status transitions. A small watcher task inside `start()` plumbs the two `watch::Receiver`s into the broadcast. The Tauri layer (`stream_stats`) subscribes via `BridgeHandle::subscribe_stats()`, takes a stats + status snapshot on every pulse (and once at startup), and emits two events `bridge://stats` / `bridge://status` to the webview. The React hook replaces `setInterval(refresh, 1000)` with two `listen()` subscriptions; the initial `refresh()` still seeds the state for the first frame. Backend now has zero `time::sleep`-driven polling loops: the syncer is driven by Phase 0 + bootstrap + the event bus, prefetch is driven by store changes, and the UI is driven by pushes. The only remaining delays are throttling (`INTER_FOLDER_DELAY`, `INTER_REQUEST_DELAY`) and reconnect backoff, which are not polls. 166 bridge lib tests, full workspace builds, UI tsc clean.
This commit is contained in:
@@ -84,6 +84,10 @@ pub struct BridgeHandle {
|
||||
status: Arc<RwLock<BridgeStatus>>,
|
||||
shutdown_tx: Option<oneshot::Sender<()>>,
|
||||
log_tx: broadcast::Sender<String>,
|
||||
/// Fires whenever something `stats()` would surface has changed —
|
||||
/// mail count, ws state, uptime tick on start/stop. Lets the UI replace
|
||||
/// its 1s poll with a push subscription.
|
||||
stats_dirty_tx: broadcast::Sender<()>,
|
||||
started_at: Option<std::time::Instant>,
|
||||
store: Option<Arc<MailStore>>,
|
||||
task: Option<tokio::task::JoinHandle<()>>,
|
||||
@@ -94,10 +98,12 @@ pub struct BridgeHandle {
|
||||
impl BridgeHandle {
|
||||
pub fn new() -> Self {
|
||||
let (log_tx, _) = broadcast::channel(256);
|
||||
let (stats_dirty_tx, _) = broadcast::channel(16);
|
||||
Self {
|
||||
status: Arc::new(RwLock::new(BridgeStatus::Stopped)),
|
||||
shutdown_tx: None,
|
||||
log_tx,
|
||||
stats_dirty_tx,
|
||||
started_at: None,
|
||||
store: None,
|
||||
task: None,
|
||||
@@ -113,6 +119,13 @@ impl BridgeHandle {
|
||||
self.log_tx.clone()
|
||||
}
|
||||
|
||||
/// Subscribe to "stats dirty" pulses. The receiver gets a `()` every
|
||||
/// time something that `stats()` would surface has changed; the UI calls
|
||||
/// `stats()` on each pulse instead of polling on a timer.
|
||||
pub fn subscribe_stats(&self) -> broadcast::Receiver<()> {
|
||||
self.stats_dirty_tx.subscribe()
|
||||
}
|
||||
|
||||
pub async fn status(&self) -> BridgeStatus {
|
||||
self.status.read().await.clone()
|
||||
}
|
||||
@@ -268,6 +281,28 @@ impl BridgeHandle {
|
||||
let bus_ids_for_handler = bus_client.last_batch_ids();
|
||||
self.ws_state_rx = Some(bus_client.state());
|
||||
|
||||
// Spawn a tiny watcher BEFORE the outer task swallows the
|
||||
// observables: it turns MailStore / WsState changes into
|
||||
// `stats_dirty_tx` pulses so the UI can replace its 1s poll with
|
||||
// an event subscription. Aborts when shutdown fires.
|
||||
{
|
||||
let stats_dirty = self.stats_dirty_tx.clone();
|
||||
let mut store_watch = store.subscribe();
|
||||
let mut ws_watch = bus_client.state();
|
||||
let mut shutdown_watch = shutdown_sync_rx.clone();
|
||||
tokio::spawn(async move {
|
||||
// Initial pulse so a subscriber sees the freshly-started state.
|
||||
let _ = stats_dirty.send(());
|
||||
loop {
|
||||
tokio::select! {
|
||||
_ = store_watch.changed() => { let _ = stats_dirty.send(()); }
|
||||
_ = ws_watch.changed() => { let _ = stats_dirty.send(()); }
|
||||
_ = shutdown_watch.changed() => return,
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
let task = tokio::spawn(async move {
|
||||
let imap_tls = tls_acceptor.clone();
|
||||
let smtp_tls = tls_acceptor;
|
||||
@@ -362,8 +397,10 @@ impl BridgeHandle {
|
||||
});
|
||||
|
||||
self.task = Some(task);
|
||||
|
||||
*self.status.write().await = BridgeStatus::Running;
|
||||
self.emit_log("Bridge is running");
|
||||
let _ = self.stats_dirty_tx.send(()); // status transition
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -376,6 +413,9 @@ impl BridgeHandle {
|
||||
}
|
||||
self.started_at = None;
|
||||
self.ws_state_rx = None;
|
||||
// Final pulse so the UI re-reads `stats()` (now reporting Stopped /
|
||||
// zero uptime / zero mails) without waiting for a poll.
|
||||
let _ = self.stats_dirty_tx.send(());
|
||||
}
|
||||
|
||||
fn emit_log(&self, msg: &str) {
|
||||
|
||||
@@ -12,8 +12,6 @@ use crate::tuta::{FolderInfo, MailBackend};
|
||||
|
||||
const INTER_REQUEST_DELAY: Duration = Duration::from_millis(150);
|
||||
const INTER_FOLDER_DELAY: Duration = Duration::from_millis(300);
|
||||
/// Pause between background body-prefetch sweeps.
|
||||
const PREFETCH_INTERVAL: Duration = Duration::from_secs(30);
|
||||
const MAX_RETRIES: u32 = 3;
|
||||
|
||||
#[derive(Clone)]
|
||||
@@ -384,14 +382,19 @@ pub async fn run_syncer(
|
||||
info!("Mail syncer shutting down");
|
||||
}
|
||||
|
||||
/// Slow loop: progressively prefetch mail bodies in the background, on its own
|
||||
/// cadence so it never delays `list_sync_loop`.
|
||||
/// Event-driven background prefetch. Wakes whenever the `MailStore`'s
|
||||
/// generation counter changes (a new mail arrived, a sync finished, …) and
|
||||
/// catches up any folder that still has bodies missing. When the store is
|
||||
/// quiet — every Mail has its `.eml` cached, no folder mutations in flight —
|
||||
/// the loop sleeps indefinitely on the watch channel; no timer-driven
|
||||
/// polling.
|
||||
async fn prefetch_loop(
|
||||
store: &MailStore,
|
||||
local_store: &LocalStore,
|
||||
backend: &dyn MailBackend,
|
||||
mut shutdown: watch::Receiver<bool>,
|
||||
) {
|
||||
let mut store_watch = store.subscribe();
|
||||
loop {
|
||||
if *shutdown.borrow() {
|
||||
return;
|
||||
@@ -403,8 +406,11 @@ async fn prefetch_loop(
|
||||
prefetch_details(store, local_store, backend, &folder).await;
|
||||
}
|
||||
|
||||
// Wait for the store to change again (or shutdown). `changed()` on a
|
||||
// freshly-borrowed receiver returns immediately if a bump happened
|
||||
// during the prefetch pass above, so we never miss a generation.
|
||||
tokio::select! {
|
||||
_ = tokio::time::sleep(PREFETCH_INTERVAL) => {}
|
||||
_ = store_watch.changed() => {}
|
||||
_ = shutdown.changed() => return,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -18,6 +18,7 @@ fn main() {
|
||||
|
||||
let handle = BridgeHandle::new();
|
||||
let log_rx = handle.subscribe_logs();
|
||||
let stats_rx = handle.subscribe_stats();
|
||||
let shared = Arc::new(Mutex::new(handle));
|
||||
|
||||
tauri::Builder::default()
|
||||
@@ -40,6 +41,12 @@ fn main() {
|
||||
stream_logs(app_handle, log_rx).await;
|
||||
});
|
||||
|
||||
let app_handle = app.handle().clone();
|
||||
let stats_state = app.state::<BridgeState>().inner().clone();
|
||||
tauri::async_runtime::spawn(async move {
|
||||
stream_stats(app_handle, stats_rx, stats_state).await;
|
||||
});
|
||||
|
||||
let state = app.state::<BridgeState>().inner().clone();
|
||||
tauri::async_runtime::spawn(async move {
|
||||
auto_start(state).await;
|
||||
@@ -75,6 +82,36 @@ async fn auto_start(state: Arc<Mutex<BridgeHandle>>) {
|
||||
}
|
||||
}
|
||||
|
||||
async fn stream_stats(
|
||||
app: tauri::AppHandle,
|
||||
mut rx: tokio::sync::broadcast::Receiver<()>,
|
||||
state: BridgeState,
|
||||
) {
|
||||
use tauri::Emitter;
|
||||
// Emit a single snapshot covering both bridge status and stats. Status
|
||||
// transitions (start / stop) and stats changes (new mail, ws state) all
|
||||
// pulse the same channel, so the UI replaces its periodic poll with one
|
||||
// listen per topic.
|
||||
async fn emit_snapshot(app: &tauri::AppHandle, state: &BridgeState) {
|
||||
let handle = state.lock().await;
|
||||
let status = handle.status().await;
|
||||
let stats = handle.stats().await;
|
||||
drop(handle);
|
||||
let _ = app.emit("bridge://stats", &stats);
|
||||
let _ = app.emit("bridge://status", &status);
|
||||
}
|
||||
|
||||
emit_snapshot(&app, &state).await;
|
||||
loop {
|
||||
match rx.recv().await {
|
||||
Ok(()) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
|
||||
emit_snapshot(&app, &state).await;
|
||||
}
|
||||
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn stream_logs(
|
||||
app: tauri::AppHandle,
|
||||
mut rx: tokio::sync::broadcast::Receiver<String>,
|
||||
|
||||
@@ -1,10 +1,9 @@
|
||||
import { useState, useEffect, useCallback, useRef } from "react";
|
||||
import { useState, useEffect, useCallback } from "react";
|
||||
import { invoke } from "@tauri-apps/api/core";
|
||||
import { listen } from "@tauri-apps/api/event";
|
||||
import type { Config, BridgeStatus, BridgeStats } from "../types";
|
||||
|
||||
const MAX_LOG_LINES = 500;
|
||||
const POLL_INTERVAL = 1000;
|
||||
|
||||
export function useBridge() {
|
||||
const [config, setConfig] = useState<Config | null>(null);
|
||||
@@ -18,7 +17,6 @@ export function useBridge() {
|
||||
const [bridgePassword, setBridgePassword] = useState<string | null>(null);
|
||||
const [logs, setLogs] = useState<string[]>([]);
|
||||
const [loading, setLoading] = useState(false);
|
||||
const pollRef = useRef<ReturnType<typeof setInterval> | null>(null);
|
||||
|
||||
const refresh = useCallback(() => {
|
||||
invoke<BridgeStatus>("get_status").then(setStatus);
|
||||
@@ -31,9 +29,13 @@ export function useBridge() {
|
||||
invoke<string | null>("get_bridge_password").then(setBridgePassword);
|
||||
refresh();
|
||||
|
||||
pollRef.current = setInterval(refresh, POLL_INTERVAL);
|
||||
// The bridge pushes `bridge://stats` and `bridge://status` whenever
|
||||
// anything changes (mail count, ws state, start/stop). No setInterval.
|
||||
const unlistenStats = listen<BridgeStats>("bridge://stats", (e) => setStats(e.payload));
|
||||
const unlistenStatus = listen<BridgeStatus>("bridge://status", (e) => setStatus(e.payload));
|
||||
return () => {
|
||||
if (pollRef.current) clearInterval(pollRef.current);
|
||||
unlistenStats.then((fn) => fn());
|
||||
unlistenStatus.then((fn) => fn());
|
||||
};
|
||||
}, [refresh]);
|
||||
|
||||
|
||||
Reference in New Issue
Block a user