Files
buzz/desktop/src-tauri/src/shutdown.rs
Tyler 2d280376ad perf(desktop): persist channel heads, collapse thread reads and reply sends (#6572)
## Summary

Lands the build-now items from the desktop latency plan
(#ui-performance-deep-dive) as one change. Every perceived-latency hot
path a user hits on launch, channel open, thread open, and reply send
drops one or more round trips.

**A1 — persisted channel heads (the big one).** Native WAL SQLite cache
(`desktop/src-tauri/src/channel_head_cache.rs`) keyed by `{pubkey,
relayUrl}` scope, 32 rows/scope LRU, 1 MiB per-row drop cap,
schema-version reset, corrupt-row tolerance, checkpointed on shutdown.
Three blocking-pool commands: `channel_head_cache_load` / `_store` /
`_clear`. On the renderer side, `CommunityQueryProvider` kicks off
hydration of up to 12 heads when it constructs the query client — the
app, splash and relay preconnect mount immediately; only
`useChannelMessagesQuery` awaits the seed (`channelHeadHydration`), then
consumes a one-shot hydrated gate so a hydrated channel pays **zero**
`get_channel_window` calls on mount and exactly **one** on the
post-subscription refresh, whose response replaces page zero wholesale.
That refresh fires whether live-subscription setup succeeds or fails,
and is sequenced behind hydration so it is always a distinct
authoritative fetch (see Review follow-ups). Bounds-only persisted heads
(zero rows) are not hydrated and take the cold loading path. The
timeline loading latch recognizes native-hydrated rows as restart-safe
so they paint immediately instead of holding a skeleton. The cache is a
paint accelerator only — the relay response is always authoritative.
Replaces the legacy localStorage `messageSnapshot.ts` (removed, -401
lines).

Kill switch: `VITE_BUZZ_CHANNEL_HEAD_CACHE=off` at build time or
`localStorage["buzz-channel-head-cache"] = "off"` at runtime. Cache is
cleared on community removal and scoped per identity, so a replaced
signer never sees the previous identity's rows.

**B1 — thread aux in one response.** Relay thread filters accept
`include_aux`; the bridge appends the same authorized two-hop
reactions/edits/deletions closure a channel window gets
(`build_aux_query` shared with the window path). Renderer
`useThreadReplies` drops its two follow-up aux fetches. `next_cursor` is
computed from reply-kind rows only since aux rows are unpaged.
Documented in `docs/bridge-channel-window.md`. Thread queries keep
`staleTime: 0` (`bcfe04e2f`): an earlier revision raised it to 30s,
which CI's `thread-unread.spec.ts` caught — once the user leaves a
channel, the live subscription stops feeding that thread's cache, so a
reopen must always take the (now single) authoritative read.

**B2 — cached root on reply send.** `send_channel_message` gains
`root_event_id`; when the renderer already holds the parent (channel or
thread cache) it passes the NIP-10 root, and native signs without the
relay round trip that `resolve_thread_ref` used to make. Strict hex
parse; `root_event_id` requires `parent_event_id`; absent root falls
back to the existing relay resolution. The renderer never sends a
guessed root.

**B4** general HTTP pool idle 10s→300s, max idle per host 1→2. **B5**
relay preconnect fires as soon as identity is ready instead of waiting
for `requestIdleCallback`. One e2e test (`relay-reconnect.spec.ts`
"service restart close resets accumulated backoff") had been relying on
the idle-callback batching to skip past its own seeded dial failures
before the channel list painted; `8133d70bb` makes it wait for the
connected state instead (test-only, still fails with the 1012 backoff
reset disabled). **B6** profile freshness 60s→10 min (both the in-memory
entry check and the query `staleTime`). Tradeoff: another user's
display-name/avatar edit can take up to 10 min to propagate to a client
that already holds their profile (relay reconnect refetches
`users-batch` but resolves from the still-fresh per-pubkey entry); your
own edits still evict the entry immediately (`evictUsersBatchEntries` in
`useUpdateProfileMutation`).

### Related issue
Follows #6456/#6457/#6459/#6460 (already merged). #6455 is the
measurement instrument and is intentionally not folded in. No duplicate
PR found.

### Review follow-ups
Addressing Carl's reviews
[5001114109](https://github.com/block/buzz/pull/6572#pullrequestreview-5001114109)
and
[5002596542](https://github.com/block/buzz/pull/6572#pullrequestreview-5002596542),
each pushed as new commits (no rebase):

- `4f06b7770` fix(desktop): mount app while channel heads hydrate;
always revalidate — provider no longer gates children on the cache load;
`refreshAfterSubscribe` runs on subscribe failure too; bounds-only heads
skipped at seed; seed merges into an existing window store. +3 tests.
- `35834cb31` fix(relay): drain aux closure hops across the page clamp —
`query_all_pages` walks the `(created_at, id)` keyset via
`until`/`before_id` until a short page (`AUX_PAGE_LIMIT` =
`DEFAULT_MAX_PAGE_LIMIT`, `AUX_MAX_PAGES` = 64 warn+truncate) so
one-shot `limit: 1000` newest-first no longer drops the oldest
edits/deletions. +3 tests; `docs/bridge-channel-window.md` updated.
- `db21b0531` merge of `origin/main` `e23632941` (#6558, #6312 — no
overlap).
- `5a5566c0f` fix(desktop): sequence post-subscribe refresh behind
channel head hydration — `refreshChannelWindowMessages` awaits
`channelHeadHydration()` and, for a hydration-seeded query (`data !==
undefined && dataUpdatedAt === 0`), the in-flight snapshot fetch before
invalidating. Without this, a subscription that settles before the
SQLite load invalidated a data-less in-flight query; TanStack dedupes
that onto the existing fetch (`query-core` `fetch()` only cancels when
`state.data` exists), which returned the seeded snapshot — 0
authoritative fetches. Regression test reproduces Carl's exact ordering
(fails at `35834cb31` with 0 calls), plus a cold-channel guard that the
fix does not double-fetch.
- `b129231c8` fix(desktop): let concurrent post-hydration refreshes
share one window fetch — found independently by Max and Wren reviewing
`5a5566c0f`: subscribe settlement + reconnect both wake on the same
snapshot promise and both invalidate; the second (default
`cancelRefetch: true`) cancelled and replaced the first authoritative
fetch (3 queryFn calls, not 2, and the cancelled Tauri invoke still hits
the relay). The seeded branch now invalidates with `cancelRefetch:
false` so a second waker joins the in-flight fetch; cold/warm keep the
default (`test_canceled_stale_fetch_cannot_overwrite_catch_up_window`
relies on it). Concurrent regression test fails at `5a5566c0f` with 3.

### Testing
At `b129231c8` (PR head; verified in one shell with `git rev-parse HEAD`
= `b129231c8`): `pnpm check`, `tsc --noEmit`, desktop unit 5,393 / 0,
Playwright `boot-splash` + `channel-head-restart` + `relay-reconnect` +
`relay-reconnect-affordance` + `thread-unread` 34 / 34 on a fresh
`build:e2e`, pre-push hooks green.

At `5a5566c0f`: `pnpm check`, `tsc --noEmit`, desktop unit 5,392 / 0,
Playwright `boot-splash` + `channel-head-restart` + `relay-reconnect` +
`relay-reconnect-affordance` + `thread-unread` 34 / 34 on a fresh
`build:e2e`, pre-push hooks green.

At `35834cb31`: desktop unit 5,390 / 0; `cargo test -p buzz-relay --lib`
910 / 0; fmt + clippy `-D warnings` clean; Playwright 32 / 32 (same
specs minus affordance); GitHub CI green on every job except Smoke (3)
(unrelated project-review row-count + messaging timing flake, per Carl)
and Unit Tests (sherpa cache skeleton, below).

Earlier, all at `8133d70bb` (this PR head is `0c492366d` = 8133d70bb + a
comments-only commit correcting two `profile/hooks.ts` freshness
comments from 60s to 10 min; pre-push desktop check/typecheck/test
5,387/0 re-ran at 0c492366d) in one shell; `origin/main` = `040b203f7`
at PR open, since moved to `4baccd539` (#6558, mobile only — zero file
overlap, `git merge-tree` clean):

- `just desktop-test` — 5,387 passed / 0 failed (includes new hook-level
call-count test: cold = 1, stale-prefetched = 1, hydrated = 0 on mount
then 1 on invalidate with wholesale replacement)
- Playwright smoke `relay-reconnect.spec.ts` + `thread-unread.spec.ts` +
`channel-head-restart.spec.ts` — 30/30 (thread-unread was 8/13 at
`7acbf951b`; relay-reconnect was 15/16 at `bcfe04e2f`). The restart spec
persists a head, reloads into a fresh mock relay with the head fetch
held 5s, asserts the persisted row paints within 2s, exactly one
`get_channel_window` after open, and the stale row is removed when the
authoritative page lands.
- `pnpm typecheck`, `pnpm check` — clean

At `7acbf951b` (everything except the two-line `useThreadReplies.ts`
staleTime revert and the test-only `relay-reconnect.spec.ts` change),
also green in one shell:
- `just desktop-tauri-test` — 2,859 passed / 0 failed across the
workspace (channel_head_cache: wire shape, LRU+caps, schema reset,
corrupt-row skip)
- `just test-unit` — 632 passed (buzz-core/auth); `cargo test -p
buzz-relay --lib` — 908 passed / 0 failed
- `just check` components: fmt-check, clippy, desktop-check,
desktop-typecheck, desktop-tauri-fmt-check, desktop-tauri-clippy,
web-check, mobile-check, file-size-check — all green
- `just desktop-build`, `web-build`, `desktop-tauri-check`,
`mobile-test` (1,661 passed) — all green

CI note: the "Unit Tests" job goes red on this PR and on `main` whenever
it hits a poisoned `rust-cache` entry (an empty-directory skeleton of
`target/sherpa-onnx-prebuilt` that `sherpa-onnx-sys` build.rs trusts),
surfacing as `could not find native static library sherpa-onnx-c-api` in
`buzz-voice` — a crate this PR doesn't touch. Deleting the cache entry
and rerunning turned the job green at `0c492366d` (28/28); it re-poisons
on the next `main` push until the workflow clears that directory after
cache restore.

Reviewed in-channel by Wren (9 / 9 / 9.5) and Eva (9 / 9 / 9), and
line-by-line by me before opening; the staleTime fix re-verified by Wren
and me independently; the relay-reconnect test fix bisected and verified
by me.

---------

Signed-off-by: Perci <5a968df9a7494b4e019b9ecf739e088ba61097b4312124e9a88ae5b42e3f5f3e@buzz.block.builderlab.xyz>
Signed-off-by: Max <d8473ee32b973aa31a21a65adddcc4b69cc2a8a4dee8121ecd51926e0cddbc02@buzz.block.builderlab.xyz>
Signed-off-by: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz>
Signed-off-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz>
Co-authored-by: Perci <5a968df9a7494b4e019b9ecf739e088ba61097b4312124e9a88ae5b42e3f5f3e@buzz.block.builderlab.xyz>
Co-authored-by: Max <d8473ee32b973aa31a21a65adddcc4b69cc2a8a4dee8121ecd51926e0cddbc02@buzz.block.builderlab.xyz>
Co-authored-by: Wren <5217c5c2f7bfb4333e46d17c98a9255a52dadee18dcd43a43536b95e6776dfa0@buzz.block.builderlab.xyz>
Co-authored-by: Meli <5aaa86bce934fc3445fc254aab560a40923f10252f92107e665073dede0e04d3@buzz.block.builderlab.xyz>
2026-08-24 10:11:07 -04:00

283 lines
10 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

use tauri::Manager;
use crate::app_state::AppState;
use crate::managed_agents::{
self, kill_stale_tracked_processes, load_managed_agents, save_managed_agents,
sync_managed_agent_processes, BackendKind,
};
use crate::{prevent_sleep, util};
pub(crate) fn is_restart_request(code: Option<i32>) -> bool {
code == Some(tauri::RESTART_EXIT_CODE)
}
pub(crate) fn shut_down_app(app: &tauri::AppHandle, shutdown_done: &std::sync::atomic::AtomicBool) {
use std::sync::atomic::Ordering;
app.state::<AppState>()
.shutdown_started
.store(true, Ordering::SeqCst);
if !shutdown_done.swap(true, Ordering::SeqCst) {
prevent_sleep::release(&app.state::<AppState>().prevent_sleep);
crate::observed_unread::flush(app);
crate::channel_head_cache::flush(app);
app.state::<crate::terminal_runtime::TerminalSessions>()
.shutdown_all();
if let Err(error) = shutdown_managed_agents(app) {
eprintln!("buzz-desktop: failed to stop managed agents: {error}");
}
#[cfg(feature = "mesh-llm")]
shutdown_mesh_runtime(app);
}
}
/// Install SIGINT/SIGTERM/SIGHUP cleanup on ctrlc's dedicated handler thread.
#[cfg(unix)]
pub(crate) fn install_signal_handler(
app: tauri::AppHandle,
shutdown_done: std::sync::Arc<std::sync::atomic::AtomicBool>,
) {
use std::sync::atomic::Ordering;
if let Err(error) = ctrlc::set_handler(move || {
app.state::<AppState>()
.shutdown_started
.store(true, Ordering::SeqCst);
if !shutdown_done.swap(true, Ordering::SeqCst) {
app.state::<crate::terminal_runtime::TerminalSessions>()
.shutdown_all();
let _ = shutdown_managed_agents(&app);
#[cfg(feature = "mesh-llm")]
shutdown_mesh_runtime(&app);
}
#[cfg(all(feature = "mesh-llm", target_os = "macos"))]
hard_exit_after_mesh_shutdown();
#[cfg(not(all(feature = "mesh-llm", target_os = "macos")))]
std::process::exit(0);
}) {
eprintln!("buzz-desktop: failed to register signal handler: {error}");
}
}
#[cfg(all(feature = "mesh-llm", target_os = "macos"))]
fn updated_macos_binary(current_binary: &std::path::Path) -> Option<std::path::PathBuf> {
let macos_directory = current_binary.parent()?;
if macos_directory.file_name()? != "MacOS" {
return None;
}
let contents_directory = macos_directory.parent()?;
if contents_directory.file_name()? != "Contents" {
return None;
}
let info_plist =
plist::from_file::<_, plist::Dictionary>(contents_directory.join("Info.plist")).ok()?;
let binary_name = info_plist.get("CFBundleExecutable")?.as_string()?;
Some(macos_directory.join(binary_name))
}
#[cfg(all(feature = "mesh-llm", target_os = "macos"))]
pub(crate) fn relaunch_after_mesh_shutdown(app: &tauri::AppHandle) -> ! {
use std::process::Command;
tauri_plugin_single_instance::destroy(app);
let env = app.env();
match tauri::process::current_binary(&env) {
Ok(current_binary) => {
let binary = updated_macos_binary(&current_binary).unwrap_or(current_binary);
if let Err(error) = Command::new(binary)
.args(env.args_os.iter().skip(1))
.spawn()
{
eprintln!("buzz-desktop: failed to relaunch app: {error}");
}
}
Err(error) => eprintln!("buzz-desktop: failed to locate app for relaunch: {error}"),
}
hard_exit_after_mesh_shutdown();
}
#[cfg(all(feature = "mesh-llm", target_os = "macos"))]
pub(crate) fn hard_exit_after_mesh_shutdown() -> ! {
// SAFETY: all Buzz-managed subprocesses and the embedded Mesh runtime have
// been stopped. `_exit` intentionally skips only process-global C++
// destructors and buffered stdio; no application state remains observable.
unsafe { libc::_exit(0) }
}
#[cfg(feature = "mesh-llm")]
pub(crate) fn shutdown_mesh_runtime(app: &tauri::AppHandle) {
let app = app.clone();
let (tx, rx) = std::sync::mpsc::channel();
tauri::async_runtime::spawn(async move {
let state = app.state::<AppState>();
let runtime = state.mesh_llm_runtime.lock().await.take();
let result = match runtime {
Some(runtime) => runtime.stop().await,
None => Ok(()),
};
let _ = tx.send(result);
});
match rx.recv_timeout(std::time::Duration::from_secs(5)) {
Ok(Ok(())) => {}
Ok(Err(error)) => eprintln!("buzz-desktop: failed to stop Mesh runtime: {error}"),
Err(error) => eprintln!("buzz-desktop: timed out stopping Mesh runtime: {error}"),
}
}
pub(crate) fn shutdown_managed_agents(app: &tauri::AppHandle) -> Result<(), String> {
let state = app.state::<AppState>();
let _restore_transition = state
.managed_agent_runtime_transition
.lock()
.map_err(|error| error.to_string())?;
let _store_guard = state
.managed_agents_store_lock
.lock()
.map_err(|error| error.to_string())?;
let mut records = load_managed_agents(app)?;
let mut runtimes = state
.managed_agent_processes
.lock()
.map_err(|error| error.to_string())?;
let (mut changed, _exited) = sync_managed_agent_processes(
&mut records,
&mut runtimes,
&managed_agents::current_instance_id(app),
);
changed |= kill_stale_tracked_processes(
&mut records,
&runtimes,
&managed_agents::current_instance_id(app),
);
// Stop all tracked agents. Send SIGTERM to all process
// groups first, then wait for exits in parallel to avoid serial 1s waits.
struct AgentToStop {
idx: usize,
pid: u32,
runtime: Option<managed_agents::ManagedAgentPairRuntime>,
}
let mut to_stop: Vec<AgentToStop> = Vec::new();
for (idx, record) in records.iter().enumerate() {
if record.backend != BackendKind::Local {
continue;
}
// Drain every tracked pair for this record, not just the first — an
// agent can run one harness per community, and each pair gets the
// graceful SIGTERM → 2s wait → SIGKILL fan-out with a stop log
// marker, instead of falling through to the orphan sweep's 200ms
// grace below.
for key in managed_agents::managed_agent_runtime_keys(&runtimes, &record.pubkey) {
let runtime = runtimes.remove(&key);
let Some(pid) = runtime
.as_ref()
.map(|rt| rt.child.id())
.or(record.runtime_pid)
else {
continue;
};
to_stop.push(AgentToStop { idx, pid, runtime });
}
}
if !to_stop.is_empty() {
changed = true;
// Fan-out: send SIGTERM to all process groups at once.
#[cfg(unix)]
for agent in &to_stop {
let pgid = -(agent.pid as i32);
unsafe {
libc::kill(pgid, libc::SIGTERM);
}
}
// Wait up to 2s for all to exit, checking in a polling loop.
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
loop {
if to_stop
.iter()
.all(|a| !managed_agents::process_is_running(a.pid))
{
break;
}
if std::time::Instant::now() >= deadline {
break;
}
std::thread::sleep(std::time::Duration::from_millis(50));
}
// Fan-out: SIGKILL any survivors.
#[cfg(unix)]
for agent in &to_stop {
if managed_agents::process_is_running(agent.pid) {
let pgid = -(agent.pid as i32);
unsafe {
libc::kill(pgid, libc::SIGKILL);
}
}
}
// Reap children and update records.
for mut agent in to_stop {
if let Some(ref mut rt) = agent.runtime {
// Best-effort reap — don’t block shutdown if the child is stuck
// in uninterruptible sleep. The zombie will be cleaned up when
// our process exits and launchd reaps it.
let _ = rt.child.try_wait();
// Write log marker (best-effort).
let record = &records[agent.idx];
let _ = managed_agents::append_log_marker(
&rt.log_path,
&format!(
"=== stopped {} ({}) at {} ===",
record.name,
record.pubkey,
util::now_iso()
),
);
}
let record = &mut records[agent.idx];
record.runtime_pid = None;
record.last_stopped_at = Some(util::now_iso());
record.updated_at = util::now_iso();
record.last_exit_code = None;
record.last_error = None;
}
}
// Final sweep: kill any orphaned agent processes we have PID file receipts
// for that escaped process-group kills or weren't tracked in records.
// All tracked PIDs have already been killed above, so pass an empty skip list.
managed_agents::sweep_orphaned_agent_processes(app, &[]);
// System-wide sweep: agent workers (goose, buzz-agent, etc.) are spawned
// in their own process groups by buzz-acp, so group-kills above only
// reach the harness, not the workers. Scan all user processes and kill any
// known agent binaries that are still running.
managed_agents::sweep_system_agent_processes(&managed_agents::current_instance_id(app), &[]);
// Dead-instance reaping: find agents belonging to Buzz instances
// whose desktop process is no longer running and reap them.
managed_agents::reap_dead_instance_agents(&managed_agents::current_instance_id(app), &[]);
if changed {
save_managed_agents(app, &records)?;
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::is_restart_request;
#[test]
fn only_tauri_restart_exit_code_requests_a_relaunch() {
assert!(is_restart_request(Some(tauri::RESTART_EXIT_CODE)));
assert!(!is_restart_request(None));
assert!(!is_restart_request(Some(0)));
}
}