perf(desktop): move the local-archive subscription into Rust

The renderer owned the whole archive path: it listed saved subscriptions,
opened a live REQ per scope, buffered inbound frames, batched them, and
handed each batch back over IPC to be written to SQLite. Every archived
frame therefore made a round trip into JS for no reason other than
history — nothing in that path is a rendering concern.

`archive::sync` now runs the whole pipeline natively. It subscribes from
the saved subscriptions, buffers, flushes on the same thresholds the JS
manager used (FLUSH_BATCH_SIZE 25 / 2000ms deadline — parity, not a
retune), archives via SQLite directly, and emits
`archive-agent-metrics-changed` when a batch actually persisted metric
rows. `archiveSyncManager.ts` is deleted.

`native_relay_client` is the shared piece underneath: one authenticated
socket per (relay, pubkey), multiplexed subscriptions, declarative
`set_subscriptions` reconciliation, and exponential reconnect backoff.
A 30s read timeout is idle, not failure, and is discriminated by error
variant rather than message text so a reworded error cannot turn every
quiet period into a reconnect storm.

A relay CLOSED drives its own recovery: per-id retry state lives next to
`open` in the connection (not in `desired`, which is reloaded from
SQLite and would resurrect a deletion), a dedicated select! deadline arm
fires the reopen and is disabled when nothing is scheduled, and the
CLOSED message is classified terminal / rate-limited / retryable with
the same prefixes the renderer used, `auth-required:` deliberately
retryable. Rate-limited arms the shared relay_admission gate and waits
max(backoff, hint); backoff is 1s→30s saturating; attempts reset on
EVENT or EOSE. Terminal suppression is per-socket by design: a
reconnect retries a terminal id once through the normal path, because
relay policy can change and one REQ per reconnect is bounded.

The sync lifecycle is owned, not raced. The renderer allocates a
monotonic lease synchronously in effect order — intent order, which IPC
completion order is not — and Rust ignores any start/stop older than
the highest (epoch, lease) mark it has seen; a stop advances the mark,
so a delayed start cannot resurrect a stopped task. Above the lease,
Rust mints a realm epoch, published atomically with minting under the
same lock that orders lifecycle calls: announcing IS what supersedes
the previous realm, so a separately-held counter would leave a window
in which a dead realm's delayed calls still win. The renderer awaits
the epoch before its first lifecycle command. Ownership is
main-window-only via the established huddleWindowChannelId() exclusion:
a huddle companion mounts the same tree in a concurrent realm, and
concurrent owners cannot be ordered by any newest-wins clock.

What stays in JS is the start gate, deliberately. Kind 24200 is
relay-ephemeral, so frames emitted before the listener opens are lost
permanently, and only the renderer knows when observer reconciliation
finished seeding 24200 into the saved subscription. The backend task is
therefore not self-starting: `useArchiveSync` starts it once
reconciliation resolves and stops it on unmount.

`RelaySession.revision` was documented as rejecting stale in-flight
reconciliation but never did — removed rather than repaired:
declarative reconciliation re-reads the desired set every pass, so
there is no generation to guard.

Tests: archive/sync_tests.rs drives the real run_sync body through a
fake IO seam (filter parity, flush thresholds, failure isolation) plus
the ownership contract (out-of-order start/stop both directions,
announcement supersedes a dead realm's delayed calls before any new
lifecycle call, publish-(epoch,0) does not lock out the announcing
realm). native_relay_client's stub-relay test completes the NIP-42
handshake over a real TCP socket, injects CLOSED with the desired set
unchanged, and proves the REQ is retried by the deadline arm — the
relay-backed #[ignore] test additionally proves the REQ shape against a
real relay. useArchiveSync.test.mjs owns the start gate, realm
ownership, and post-reload realm supersession via fresh module
instances. observer-archive-policy.spec.ts owns wiring, with payload
receipts that announce precedes start and start carries a numeric
epoch. Every claim was mutation-checked; the vacuous first drafts
(wake-masked reopen, policy re-implementation, precondition-rebuilding
supersession, same-scope no-op escape) were each caught by their
mutants and rewritten.

Includes one move-only hunk that is not archive work: the push-to-talk
global-shortcut handler moves from lib.rs into `ptt_shortcut::install`,
mirroring the existing `app_menu::install` seam, paying the 1000-line
ratchet budget in the module that owns the registration lifecycle. The
handler body is proven token-identical with a mutated-body negative
control. lib.rs is 917 lines.

Co-authored-by: Tyler Longwell <tlongwell@squareup.com>
Signed-off-by: Tyler Longwell <tlongwell@squareup.com>
This commit is contained in:
Dawn
2026-08-15 21:38:48 -04:00
committed by Eva
parent 9b5a82ab9a
commit 64144d2f4f
18 changed files with 3051 additions and 1667 deletions

View File

@@ -1097,6 +1097,7 @@ dependencies = [
"buzz-sdk",
"buzz-terminal",
"buzz-voice",
"buzz-ws-client",
"bytes",
"bzip2 0.6.1",
"chrono",
@@ -1253,6 +1254,20 @@ dependencies = [
"tokenizers",
]
[[package]]
name = "buzz-ws-client"
version = "0.1.0"
dependencies = [
"futures-util",
"nostr 0.44.7",
"serde_json",
"thiserror 2.0.18",
"tokio",
"tokio-tungstenite 0.29.0",
"tracing",
"url",
]
[[package]]
name = "by_address"
version = "1.2.1"

View File

@@ -108,6 +108,7 @@ buzz_sdk_pkg = { package = "buzz-sdk", path = "../../crates/buzz-sdk" }
buzz_agent_pkg = { package = "buzz-agent", path = "../../crates/buzz-agent" }
buzz_voice_pkg = { package = "buzz-voice", path = "../../crates/buzz-voice" }
buzz_terminal = { package = "buzz-terminal", path = "crates/buzz-terminal" }
buzz_ws_client_pkg = { package = "buzz-ws-client", path = "../../crates/buzz-ws-client" }
portable-pty = "0.9"
iroh = { version = "1.0.2", optional = true }
mesh-llm-sdk = { git = "https://github.com/Mesh-LLM/mesh-llm.git", tag = "v0.75.1", package = "mesh-llm-sdk", default-features = false, features = ["client", "serving"], optional = true }

View File

@@ -22,6 +22,7 @@ mod metric_store;
mod pipeline;
pub mod store;
mod store_migrations;
pub mod sync;
use pipeline::{commit_archive, plan_archive, query_buckets};
@@ -150,8 +151,20 @@ pub async fn archive_events(
state: State<'_, AppState>,
candidates: Vec<ArchiveCandidate>,
) -> Result<ArchiveBatchResult, String> {
let identity_pk = identity_pubkey(&state)?;
let relay_url = relay_ws_url_with_override(&state);
archive_candidates(&state, candidates).await
}
/// The body of [`archive_events`], callable without a command invocation.
///
/// The native sync task archives through this directly: routing its batches
/// back out to the renderer just to have the renderer invoke the command would
/// reintroduce the IPC round trip the move exists to delete.
pub(crate) async fn archive_candidates(
state: &AppState,
candidates: Vec<ArchiveCandidate>,
) -> Result<ArchiveBatchResult, String> {
let identity_pk = identity_pubkey(state)?;
let relay_url = relay_ws_url_with_override(state);
let now = now_secs();
// ── Phase 1: plan (blocking SQLite) ─────────────────────────────────────
@@ -163,8 +176,7 @@ pub async fn archive_events(
.await?;
// ── Phase 2: relay queries (async) ───────────────────────────────────────
let state_ref: &AppState = &state;
let bucket_results = query_buckets(plan.buckets, state_ref).await;
let bucket_results = query_buckets(plan.buckets, state).await;
// ── Phase 3: persist (blocking SQLite) ──────────────────────────────────
let owner_keys = {
@@ -286,6 +298,7 @@ fn validate_ephemeral_frame(
#[tauri::command]
pub async fn create_save_subscription(
state: State<'_, AppState>,
sync_state: State<'_, sync::ArchiveSyncState>,
scope_type: ScopeType,
scope_value: String,
kinds: Vec<u32>,
@@ -333,7 +346,9 @@ pub async fn create_save_subscription(
&scope_value,
&kinds_json,
now,
)
)?;
sync_state.notify_subscriptions_changed().await;
Ok(())
}
/// Probe: the current user has access to `channel_id` (kind 39002 lists them).
@@ -426,6 +441,7 @@ async fn probe_event_readable(state: &AppState, event_id: &str) -> Result<(), St
#[tauri::command]
pub async fn merge_save_subscription_kinds(
state: State<'_, AppState>,
sync_state: State<'_, sync::ArchiveSyncState>,
kind: u32,
) -> Result<(), String> {
if kind > u32::from(u16::MAX) {
@@ -439,7 +455,9 @@ pub async fn merge_save_subscription_kinds(
run_archive_db_task(move |conn| {
store::merge_owner_p_kinds(conn, &identity_pk, &relay_url, &owner_pk, kind, now)
})
.await
.await?;
sync_state.notify_subscriptions_changed().await;
Ok(())
}
// ── remove_save_subscription_kind ────────────────────────────────────────────
@@ -459,6 +477,7 @@ pub async fn merge_save_subscription_kinds(
#[tauri::command]
pub async fn remove_save_subscription_kind(
state: State<'_, AppState>,
sync_state: State<'_, sync::ArchiveSyncState>,
kind: u32,
) -> Result<(), String> {
if kind > u32::from(u16::MAX) {
@@ -471,7 +490,9 @@ pub async fn remove_save_subscription_kind(
run_archive_db_task(move |conn| {
store::remove_owner_p_kind(conn, &identity_pk, &relay_url, &owner_pk, kind)
})
.await
.await?;
sync_state.notify_subscriptions_changed().await;
Ok(())
}
// ── list_save_subscriptions ──────────────────────────────────────────────────
@@ -496,12 +517,13 @@ pub async fn list_save_subscriptions(
#[tauri::command]
pub async fn delete_save_subscription(
state: State<'_, AppState>,
sync_state: State<'_, sync::ArchiveSyncState>,
scope_type: ScopeType,
scope_value: String,
) -> Result<bool, String> {
let identity_pk = identity_pubkey(&state)?;
let relay_url = relay_ws_url_with_override(&state);
run_archive_db_task(move |conn| {
let removed = run_archive_db_task(move |conn| {
store::delete_save_subscription(
conn,
&identity_pk,
@@ -510,7 +532,11 @@ pub async fn delete_save_subscription(
&scope_value,
)
})
.await
.await?;
if removed {
sync_state.notify_subscriptions_changed().await;
}
Ok(removed)
}
// ── read_archived_events ─────────────────────────────────────────────────────

View File

@@ -0,0 +1,549 @@
//! Rust archive sync task — the backend replacement for the renderer's
//! `archiveSyncManager`.
//!
//! Opens one live relay subscription per saved archive config and forwards
//! matched events to the existing archive pipeline in debounced batches. The
//! renderer no longer sees archive traffic at all: previously every matched
//! event crossed the IPC boundary twice (relay -> renderer, renderer ->
//! `archive_events`) purely to be written to a SQLite file the backend owns.
//!
//! # Start gate
//!
//! The task is NOT self-starting. Kind 24200 is relay-*ephemeral*: frames that
//! arrive before the listener opens are permanently lost, so the renderer must
//! finish observer reconciliation (which seeds kind 24200 into the owner_p
//! subscription) before any listener opens. That ordering is the whole reason
//! `useArchiveSync` gated on `observerReconciled`, and it survives the move as
//! an explicit `start_archive_sync` command issued after the same gate.
use std::{collections::HashMap, future::Future, pin::Pin, sync::Arc, time::Duration};
use nostr::JsonUtil;
use serde_json::json;
use tauri::{AppHandle, Emitter, Manager, State};
use tokio::{
sync::{mpsc, Mutex, Notify},
time::Instant,
};
use tokio_util::sync::CancellationToken;
use super::{
store::SaveSubscription, ArchiveBatchResult, ArchiveCandidate, MatchedScope, ScopeType,
};
use crate::app_state::AppState;
use crate::native_relay_client::{self, MatchedEvent, RelaySession, Subscription};
/// Flush once this many events are buffered. Parity with the renderer manager.
const FLUSH_BATCH_SIZE: usize = 25;
/// Maximum time an event waits in the buffer before being flushed.
///
/// This is a deadline measured from the FIRST buffered event, not an idle
/// timer that each arrival extends. The renderer constant was named
/// `FLUSH_IDLE_MS`, but its `scheduleFlush` returned early when a timer was
/// already pending, so a steady trickle still flushed every 2s rather than
/// never. The behavior is preserved; the name is corrected.
const FLUSH_DEADLINE: Duration = Duration::from_millis(2_000);
/// Emitted after a batch persists new agent-metric rows, so the renderer can
/// invalidate its usage queries. Replaces the in-process `notifyAgentMetrics
/// Changed()` call the manager made on the JS side of that same batch.
const AGENT_METRICS_CHANGED_EVENT: &str = "archive-agent-metrics-changed";
type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
/// Everything the sync loop needs from the outside world.
///
/// Injected rather than reached for so the loop's batching, demultiplexing,
/// and reload behavior are testable without a relay, a database, or a Tauri
/// app handle.
pub(crate) trait ArchiveSyncIo: Send + Sync + 'static {
fn list_subscriptions(&self) -> BoxFuture<'_, Result<Vec<SaveSubscription>, String>>;
fn set_subscriptions(&self, subscriptions: Vec<Subscription>) -> BoxFuture<'_, ()>;
fn archive(
&self,
candidates: Vec<ArchiveCandidate>,
) -> BoxFuture<'_, Result<ArchiveBatchResult, String>>;
fn notify_agent_metrics_changed(&self);
}
// ── Subscription planning ────────────────────────────────────────────────────
/// The relay subscription set for `subscriptions`, plus the scope each
/// subscription id maps back to when its events arrive.
///
/// The id encodes scope AND kinds, so a kinds change produces a different id:
/// the session then closes the old subscription and opens the new one instead
/// of leaving a stale filter live. Same reason the renderer keyed on both.
fn plan_subscriptions(
subscriptions: &[SaveSubscription],
) -> (Vec<Subscription>, HashMap<String, MatchedScope>) {
let mut planned = Vec::new();
let mut scopes = HashMap::new();
for sub in subscriptions {
let Some(scope_type) = parse_scope_type(&sub.scope_type) else {
eprintln!(
"buzz-desktop: archive sync: unknown scope_type {:?}, skipping",
sub.scope_type
);
continue;
};
// A malformed `kinds` column decodes as empty, matching the renderer
// decoder. The resulting filter matches nothing, which is the correct
// failure for a row we cannot interpret: archive nothing, drop nothing.
let kinds: Vec<u64> = serde_json::from_str(&sub.kinds).unwrap_or_default();
let id = subscription_id(&scope_type, &sub.scope_value, &kinds);
if scopes.contains_key(&id) {
continue;
}
planned.push(Subscription {
id: id.clone(),
filter: build_filter(&scope_type, &sub.scope_value, &kinds),
});
scopes.insert(
id,
MatchedScope {
scope_type,
scope_value: sub.scope_value.clone(),
},
);
}
(planned, scopes)
}
fn parse_scope_type(raw: &str) -> Option<ScopeType> {
match raw {
"channel_h" => Some(ScopeType::ChannelH),
"owner_p" => Some(ScopeType::OwnerP),
"referenced_e" => Some(ScopeType::ReferencedE),
_ => None,
}
}
/// `limit: 0` — live tail only. Stored events are archived by the explicit
/// backfill paths, so a non-zero limit would re-deliver history on every
/// reconnect.
fn build_filter(scope_type: &ScopeType, scope_value: &str, kinds: &[u64]) -> serde_json::Value {
let tag = match scope_type {
ScopeType::ChannelH => "#h",
ScopeType::OwnerP => "#p",
ScopeType::ReferencedE => "#e",
};
json!({ "kinds": kinds, "limit": 0, tag: [scope_value] })
}
fn subscription_id(scope_type: &ScopeType, scope_value: &str, kinds: &[u64]) -> String {
let mut sorted = kinds.to_vec();
sorted.sort_unstable();
let kinds = sorted
.iter()
.map(|k| k.to_string())
.collect::<Vec<_>>()
.join(",");
format!("archive:{}:{scope_value}:{kinds}", scope_type.as_str())
}
// ── Batching ─────────────────────────────────────────────────────────────────
/// Buffered candidates plus the deadline of the oldest one.
#[derive(Default)]
struct PendingBatch {
candidates: Vec<ArchiveCandidate>,
/// Set when the buffer goes from empty to non-empty, cleared on take. The
/// deadline belongs to the oldest buffered event, so a steady trickle of
/// arrivals cannot postpone its flush indefinitely.
deadline: Option<Instant>,
}
impl PendingBatch {
fn push(&mut self, candidate: ArchiveCandidate) {
if self.candidates.is_empty() {
self.deadline = Some(Instant::now() + FLUSH_DEADLINE);
}
self.candidates.push(candidate);
}
fn is_full(&self) -> bool {
self.candidates.len() >= FLUSH_BATCH_SIZE
}
fn take(&mut self) -> Vec<ArchiveCandidate> {
self.deadline = None;
std::mem::take(&mut self.candidates)
}
}
// ── Sync loop ────────────────────────────────────────────────────────────────
/// Drives one archive sync session until `cancel` fires.
///
/// Reload requests coalesce: `Notify::notify_one` stores at most one permit, so
/// any number of subscription changes arriving during a reload produce exactly
/// one follow-up pass — the same guarantee the renderer's single-flight
/// `reloadPending` loop provided, without the bookkeeping.
async fn run_sync<I: ArchiveSyncIo + ?Sized>(
io: &I,
reload: Arc<Notify>,
mut events: mpsc::Receiver<MatchedEvent>,
cancel: CancellationToken,
) {
let mut scopes: HashMap<String, MatchedScope> = HashMap::new();
let mut pending = PendingBatch::default();
reconcile(io, &mut scopes).await;
loop {
// `Instant::far_future()` is not public; a long sleep stands in for
// "no deadline" so the select arm can be unconditional.
let deadline = pending
.deadline
.unwrap_or_else(|| Instant::now() + Duration::from_secs(3600));
tokio::select! {
_ = cancel.cancelled() => break,
_ = reload.notified() => {
reconcile(io, &mut scopes).await;
}
_ = tokio::time::sleep_until(deadline), if pending.deadline.is_some() => {
flush(io, pending.take()).await;
}
received = events.recv() => {
let Some(event) = received else { break };
// A subscription we already closed can still have events in
// flight; without its scope we cannot assert a match, and the
// backend re-verifies scope claims anyway, so drop it.
let Some(scope) = scopes.get(&event.subscription_id) else { continue };
pending.push(ArchiveCandidate {
raw_event_json: event.event.as_json(),
matched_scope: MatchedScope {
scope_type: scope.scope_type.clone(),
scope_value: scope.scope_value.clone(),
},
});
if pending.is_full() {
flush(io, pending.take()).await;
}
}
}
}
// Buffered events are already off the relay; dropping them on shutdown
// would lose them permanently for the ephemeral scope.
flush(io, pending.take()).await;
}
/// Reloads the saved subscriptions and applies them to the session.
///
/// A failed load leaves the previous set live rather than tearing everything
/// down: a transient SQLite error must not silently stop archiving.
async fn reconcile<I: ArchiveSyncIo + ?Sized>(io: &I, scopes: &mut HashMap<String, MatchedScope>) {
let subscriptions = match io.list_subscriptions().await {
Ok(subscriptions) => subscriptions,
Err(error) => {
eprintln!("buzz-desktop: archive sync: list_save_subscriptions failed: {error}");
return;
}
};
let (planned, next_scopes) = plan_subscriptions(&subscriptions);
io.set_subscriptions(planned).await;
*scopes = next_scopes;
}
/// Awaited rather than spawned: back-pressure through the session's bounded
/// event channel is what keeps a catch-up storm from queueing unbounded
/// archive work. The renderer's fire-and-forget was a property of living in
/// an event loop it could not block, not a behavior worth porting.
async fn flush<I: ArchiveSyncIo + ?Sized>(io: &I, candidates: Vec<ArchiveCandidate>) {
if candidates.is_empty() {
return;
}
match io.archive(candidates).await {
// The backend is authoritative: a duplicate-only batch or one with no
// kind-44200 events must not invalidate usage queries.
Ok(result) if result.persisted_agent_metrics > 0 => io.notify_agent_metrics_changed(),
Ok(_) => {}
Err(error) => eprintln!("buzz-desktop: archive sync: archive_events failed: {error}"),
}
}
// ── Production wiring ────────────────────────────────────────────────────────
struct AppIo {
app: AppHandle,
session: Arc<RelaySession>,
}
impl ArchiveSyncIo for AppIo {
fn list_subscriptions(&self) -> BoxFuture<'_, Result<Vec<SaveSubscription>, String>> {
Box::pin(async move {
let state: State<'_, AppState> = self.app.state();
let identity_pk = super::identity_pubkey(&state)?;
let relay_url = crate::relay::relay_ws_url_with_override(&state);
super::run_archive_db_task(move |conn| {
super::store::list_save_subscriptions(conn, &identity_pk, &relay_url)
})
.await
})
}
fn set_subscriptions(&self, subscriptions: Vec<Subscription>) -> BoxFuture<'_, ()> {
Box::pin(async move { self.session.set_subscriptions(subscriptions).await })
}
fn archive(
&self,
candidates: Vec<ArchiveCandidate>,
) -> BoxFuture<'_, Result<ArchiveBatchResult, String>> {
Box::pin(async move {
let state: State<'_, AppState> = self.app.state();
super::archive_candidates(&state, candidates).await
})
}
fn notify_agent_metrics_changed(&self) {
let _ = self.app.emit(AGENT_METRICS_CHANGED_EVENT, ());
}
}
/// Managed handle for the running sync task.
#[derive(Default)]
pub struct ArchiveSyncState {
running: Mutex<Option<RunningSync>>,
/// Highest `(epoch, lease)` this process has seen from either command.
///
/// The renderer allocates leases synchronously in effect order, so they are
/// the app's intent order — which the IPC completion order is not. Both
/// commands ignore anything older, which is what makes a stale cleanup
/// harmless and a delayed start unable to resurrect a stopped task.
///
/// The epoch is minted here, not in the renderer, because a lease counter
/// only exists for as long as the JS realm that holds it. A renderer reload
/// (`useReloadShortcut`, `RootErrorBoundary`, `useCommunityInit`) resets the
/// counter to zero while this state persists in the Tauri process, so
/// without an epoch the first post-reload start looks older than what the
/// backend already saw and is rejected forever. Ordering lexicographically
/// on `(epoch, lease)` means a newer realm outranks the old one no matter
/// where its local counter restarted.
///
/// Every boundary the intent-order authority crosses, and why it holds:
///
/// - effect remount, same realm: leases strictly increase within a realm.
/// - IPC arrival order: the lease is minted before `invoke`, so intent
/// order is fixed before the calls can race.
/// - renderer realm reload: a new epoch from this authority outranks the
/// dead realm's, whatever its counter said.
/// - webview recreation of the owning window: same as reload.
/// - a second window: it does not participate, by ownership rule. Archive
/// sync is app-global and main-window-owned, exactly as the main window
/// remains the owner of microphone capture (see `huddle::window`).
/// Epochs order realms in time; a companion window is a second realm in
/// space, and newest-wins cannot model two concurrent owners — a
/// companion's cleanup would cancel the live main-window task. Secondary
/// realms therefore never announce and never issue lifecycle commands.
/// Any future second-realm mount must revisit this.
/// - Tauri process restart: both clocks die together, so there is nothing
/// to order against.
latest: Mutex<(u64, u64)>,
}
struct RunningSync {
/// Identity + relay this task is bound to. A start request for the same
/// scope is a no-op, so a renderer remount does not churn the socket.
scope: (String, String),
cancel: CancellationToken,
reload: Arc<Notify>,
}
impl ArchiveSyncState {
/// Wakes the sync task so it reloads saved subscriptions.
///
/// Called by the archive commands that mutate `save_subscriptions`. This
/// replaces the renderer's `onSubscriptionChange` notifier: the mutations
/// were already backend commands, so routing the signal through JS only
/// created a window where a write landed but nothing resubscribed.
pub(super) async fn notify_subscriptions_changed(&self) {
if let Some(running) = self.running.lock().await.as_ref() {
running.reload.notify_one();
}
}
/// Mints the epoch a renderer realm must hold before it may issue any
/// lifecycle command, and publishes it as the current mark in the same
/// critical section.
///
/// A realm has to obtain this *before* its archive effect runs, and the
/// renderer awaits it. If announcing were just another unawaited `invoke`
/// beside the lifecycle calls, it would race them and recreate the
/// arrival-order bug one level up — the epoch would order announcements
/// rather than realms.
///
/// Minting and publishing are one lock acquisition because announcing is
/// what supersedes the old realm. Holding the epoch counter separately —
/// so a mint could not block an in-flight lifecycle call — leaves a window
/// between mint and first use in which `latest` still names the dead realm,
/// and its delayed `start`/`stop` with any lease still wins. The new realm
/// cannot close that window itself: the reconciliation gate can keep its
/// first lifecycle call arbitrarily far behind its announcement.
///
/// The published lease is `0`, the one value that outranks every mark the
/// previous realm can hold while still sitting below this realm's own first
/// lease. Publishing higher would reject the announcing realm's own start
/// and leave sync permanently unstarted.
async fn announce(&self) -> u64 {
let mut latest = self.latest.lock().await;
let epoch = latest.0 + 1;
*latest = (epoch, 0);
epoch
}
/// Takes ownership for a start under `(epoch, lease)`, installing the task
/// when it wins. Returns false when the caller must not proceed — either
/// the mark is stale or an equivalent task is already running.
///
/// The whole ownership policy lives here rather than in the command so the
/// regression tests drive the same code production does. A test that
/// re-implemented "claim, then install" would pass against a command that
/// had stopped calling either one.
///
/// The mark must be strictly newer than anything seen: that rejects both a
/// start a newer start already superseded, and one delayed past its own
/// stop. Comparison is lexicographic on `(epoch, lease)`, so any call from
/// a superseded realm loses regardless of how far its lease counter ran.
async fn begin(
&self,
mark: (u64, u64),
scope: (String, String),
cancel: CancellationToken,
reload: Arc<Notify>,
) -> bool {
let mut latest = self.latest.lock().await;
if mark <= *latest {
return false;
}
*latest = mark;
let mut running = self.running.lock().await;
// A same-scope remount keeps its socket: reinstalling would tear down a
// healthy relay session to replace it with an identical one.
if running
.as_ref()
.is_some_and(|current| current.scope == scope)
{
return false;
}
if let Some(previous) = running.take() {
previous.cancel.cancel();
}
*running = Some(RunningSync {
scope,
cancel,
reload,
});
true
}
/// Releases ownership for a stop under `(epoch, lease)`, cancelling the
/// running task when it wins.
///
/// Equality succeeds here, unlike [`Self::begin`]: a stop is the
/// counterpart of the start that minted its lease, so its own mark is
/// exactly the case it must act on. Advancing the mark is what stops a
/// start delayed past its own cleanup from resurrecting the task.
///
/// The guard is held across the cancellation, exactly as [`Self::begin`]
/// holds it across the install. Releasing it first would reopen the very
/// window this ordering closes: a stop could clear its check, yield, let a
/// newer start install its task, and then cancel that task on resume —
/// stale cleanup stranding the newest owner. Both halves take `latest` then
/// `running`, so the two can never interleave and the order is deadlock-free.
async fn end(&self, mark: (u64, u64)) {
let mut latest = self.latest.lock().await;
if mark < *latest {
return;
}
*latest = mark;
if let Some(running) = self.running.lock().await.take() {
running.cancel.cancel();
}
}
}
/// Announce a renderer realm and obtain its epoch.
///
/// The renderer awaits this before its archive effect may issue any lifecycle
/// command; see [`ArchiveSyncState::latest`] for the boundaries this closes.
/// Only the main window announces — archive sync is app-global and
/// main-window-owned.
#[tauri::command]
pub async fn announce_archive_sync_epoch(
sync_state: State<'_, ArchiveSyncState>,
) -> Result<u64, String> {
Ok(sync_state.announce().await)
}
/// Start archive sync for the current identity.
///
/// Idempotent for the same identity + relay. Issued by the renderer only after
/// observer reconciliation completes — see the module docs for why that gate
/// cannot be moved into the backend.
///
/// `epoch` identifies the calling realm and `lease` orders this call against
/// that realm's other lifecycle calls; see [`ArchiveSyncState::latest`].
#[tauri::command]
pub async fn start_archive_sync(
app: AppHandle,
state: State<'_, AppState>,
sync_state: State<'_, ArchiveSyncState>,
epoch: u64,
lease: u64,
) -> Result<(), String> {
let keys = state.signing_keys()?;
let relay_url = crate::relay::relay_ws_url_with_override(&state);
let scope = (keys.public_key().to_hex(), relay_url.clone());
// Only cheap handles before `begin`: a start that lost its mark, or a
// same-scope remount, must not open a relay socket just to drop it again.
let cancel = CancellationToken::new();
let reload = Arc::new(Notify::new());
if !sync_state
.begin((epoch, lease), scope, cancel.clone(), Arc::clone(&reload))
.await
{
return Ok(());
}
// No NIP-OA auth tag: this is the owner's own session, authenticated as
// the identity itself, exactly like the renderer's relay client.
let (session, events) = native_relay_client::start(relay_url, keys, None);
let io = AppIo {
app: app.clone(),
session: Arc::clone(&session),
};
tauri::async_runtime::spawn(async move {
run_sync(&io, reload, events, cancel).await;
session.shutdown();
});
Ok(())
}
/// Stop archive sync. Mirrors the renderer teardown that ran when the gate
/// closed (identity change, community switch, unmount).
///
/// `(epoch, lease)` is the mark its own start allocated; a cleanup that has
/// been superseded is a no-op rather than cancelling a newer owner's task.
#[tauri::command]
pub async fn stop_archive_sync(
sync_state: State<'_, ArchiveSyncState>,
epoch: u64,
lease: u64,
) -> Result<(), String> {
sync_state.end((epoch, lease)).await;
Ok(())
}
#[cfg(test)]
#[path = "sync_tests.rs"]
mod sync_tests;

View File

@@ -0,0 +1,880 @@
//! Tests for the native archive sync loop.
//!
//! The loop is driven through the real `run_sync` body with a fake
//! [`ArchiveSyncIo`] and a real event channel, so batching, demultiplexing,
//! reload coalescing, and shutdown flush are exercised as the production task
//! runs them — not as a struct poked directly.
use super::*;
use nostr::{EventBuilder, Keys, Kind, Tag};
use std::sync::Mutex as StdMutex;
// ── Test doubles ─────────────────────────────────────────────────────────────
#[derive(Default)]
struct FakeIo {
/// Successive results for `list_subscriptions`; the last one repeats so a
/// reload that outruns the script does not panic.
listings: StdMutex<Vec<Vec<SaveSubscription>>>,
applied: StdMutex<Vec<Vec<Subscription>>>,
batches: StdMutex<Vec<Vec<ArchiveCandidate>>>,
/// What `archive` returns; drives the notify-on-metrics assertion.
persisted_agent_metrics: StdMutex<u32>,
archive_fails: StdMutex<bool>,
metrics_notifications: StdMutex<u32>,
}
impl FakeIo {
fn with_listings(listings: Vec<Vec<SaveSubscription>>) -> Self {
Self {
listings: StdMutex::new(listings),
..Default::default()
}
}
fn applied(&self) -> Vec<Vec<Subscription>> {
self.applied.lock().unwrap().clone()
}
/// Flattened candidates in delivery order, as `(scope_value, event_id)`.
fn archived(&self) -> Vec<Vec<String>> {
self.batches
.lock()
.unwrap()
.iter()
.map(|batch| {
batch
.iter()
.map(|c| c.matched_scope.scope_value.clone())
.collect()
})
.collect()
}
}
impl ArchiveSyncIo for FakeIo {
fn list_subscriptions(&self) -> BoxFuture<'_, Result<Vec<SaveSubscription>, String>> {
Box::pin(async move {
let mut listings = self.listings.lock().unwrap();
if listings.is_empty() {
return Ok(Vec::new());
}
if listings.len() == 1 {
return Ok(listings[0].clone());
}
Ok(listings.remove(0))
})
}
fn set_subscriptions(&self, subscriptions: Vec<Subscription>) -> BoxFuture<'_, ()> {
Box::pin(async move {
self.applied.lock().unwrap().push(subscriptions);
})
}
fn archive(
&self,
candidates: Vec<ArchiveCandidate>,
) -> BoxFuture<'_, Result<ArchiveBatchResult, String>> {
Box::pin(async move {
self.batches.lock().unwrap().push(candidates);
if *self.archive_fails.lock().unwrap() {
return Err("archive failed".to_string());
}
Ok(ArchiveBatchResult {
persisted: 0,
persisted_agent_metrics: *self.persisted_agent_metrics.lock().unwrap(),
dropped: 0,
})
})
}
fn notify_agent_metrics_changed(&self) {
*self.metrics_notifications.lock().unwrap() += 1;
}
}
/// Yields until `condition` holds, then returns; fails the test if it never
/// does. An unbounded spin turns a broken flush into a HUNG test instead of a
/// failing one — and under a paused clock it also starves tokio's auto-advance,
/// so the deadline that would have masked the bug never even fires.
async fn wait_for(label: &str, mut condition: impl FnMut() -> bool) {
for _ in 0..10_000 {
if condition() {
return;
}
tokio::task::yield_now().await;
}
panic!("timed out waiting for {label}");
}
fn saved(scope_type: &str, scope_value: &str, kinds: &str) -> SaveSubscription {
SaveSubscription {
identity_pubkey: "owner".into(),
relay_url: "wss://relay.test".into(),
scope_type: scope_type.into(),
scope_value: scope_value.into(),
kinds: kinds.into(),
created_at: 0,
}
}
fn matched(subscription_id: &str) -> MatchedEvent {
let event = EventBuilder::new(Kind::Custom(9), "hello")
.tags([Tag::parse(vec!["h", "channel-a"]).unwrap()])
.sign_with_keys(&Keys::generate())
.unwrap();
MatchedEvent {
subscription_id: subscription_id.to_string(),
event: Box::new(event),
}
}
/// Runs `run_sync` on a task, handing back the controls the tests drive it
/// with. Every test cancels and joins, so a loop that fails to observe
/// cancellation hangs the test rather than passing silently.
fn spawn_sync(
io: Arc<FakeIo>,
) -> (
mpsc::Sender<MatchedEvent>,
Arc<Notify>,
CancellationToken,
tokio::task::JoinHandle<()>,
) {
let (tx, rx) = mpsc::channel(64);
let reload = Arc::new(Notify::new());
let cancel = CancellationToken::new();
let handle = {
let io = Arc::clone(&io);
let reload = Arc::clone(&reload);
let cancel = cancel.clone();
tokio::spawn(async move { run_sync(io.as_ref(), reload, rx, cancel).await })
};
(tx, reload, cancel, handle)
}
async fn stop(cancel: CancellationToken, handle: tokio::task::JoinHandle<()>) {
cancel.cancel();
handle.await.expect("sync task panicked");
}
// ── Filter construction ──────────────────────────────────────────────────────
#[test]
fn filters_match_the_renderer_shape_for_every_scope() {
// Verbatim parity with `buildFilter` in archiveSyncManager.ts: the tag key
// per scope and `limit: 0` are the contract with the relay, and a wrong
// tag key silently archives nothing.
let (planned, scopes) = plan_subscriptions(&[
saved("channel_h", "channel-a", "[9,40002]"),
saved("owner_p", "owner-pk", "[24200]"),
saved("referenced_e", "event-id", "[1]"),
]);
let filters: Vec<_> = planned.iter().map(|s| s.filter.clone()).collect();
assert_eq!(
filters,
vec![
json!({ "kinds": [9, 40002], "limit": 0, "#h": ["channel-a"] }),
json!({ "kinds": [24200], "limit": 0, "#p": ["owner-pk"] }),
json!({ "kinds": [1], "limit": 0, "#e": ["event-id"] }),
]
);
assert_eq!(scopes.len(), 3);
let scope = &scopes[&planned[0].id];
assert_eq!(scope.scope_type, ScopeType::ChannelH);
assert_eq!(scope.scope_value, "channel-a");
}
#[test]
fn subscription_id_changes_when_kinds_change() {
// The id doubles as the relay subscription id, so a kinds change MUST
// produce a different one — otherwise the session sees the same id with a
// new filter and the old filter can stay live.
let (before, _) = plan_subscriptions(&[saved("channel_h", "channel-a", "[9]")]);
let (after, _) = plan_subscriptions(&[saved("channel_h", "channel-a", "[9,40002]")]);
assert_ne!(before[0].id, after[0].id);
}
#[test]
fn subscription_id_is_stable_across_kind_ordering() {
// Same set written in a different order is the same subscription; without
// the sort it would churn the socket on every reload.
let (a, _) = plan_subscriptions(&[saved("channel_h", "channel-a", "[40002,9]")]);
let (b, _) = plan_subscriptions(&[saved("channel_h", "channel-a", "[9,40002]")]);
assert_eq!(a[0].id, b[0].id);
}
#[test]
fn unknown_scope_type_is_skipped_not_guessed() {
let (planned, scopes) = plan_subscriptions(&[
saved("wat", "x", "[9]"),
saved("channel_h", "channel-a", "[9]"),
]);
assert_eq!(planned.len(), 1);
assert_eq!(scopes.len(), 1);
assert_eq!(scopes[&planned[0].id].scope_value, "channel-a");
}
#[test]
fn malformed_kinds_column_yields_a_matchless_filter() {
// Mirrors the renderer decoder: a row we cannot interpret archives
// nothing rather than subscribing to everything.
let (planned, _) = plan_subscriptions(&[saved("channel_h", "channel-a", "not json")]);
assert_eq!(
planned[0].filter,
json!({ "kinds": [], "limit": 0, "#h": ["channel-a"] })
);
}
#[test]
fn duplicate_rows_produce_one_subscription() {
let (planned, _) = plan_subscriptions(&[
saved("channel_h", "channel-a", "[9]"),
saved("channel_h", "channel-a", "[9]"),
]);
assert_eq!(planned.len(), 1);
}
// ── Loop behavior ────────────────────────────────────────────────────────────
#[tokio::test]
async fn subscribes_to_saved_configs_on_start() {
let io = Arc::new(FakeIo::with_listings(vec![vec![saved(
"channel_h",
"channel-a",
"[9]",
)]]));
let (_tx, _reload, cancel, handle) = spawn_sync(Arc::clone(&io));
// The first reconcile races the cancel; wait for it to land.
wait_for("initial subscribe", || !io.applied().is_empty()).await;
assert_eq!(io.applied()[0].len(), 1);
stop(cancel, handle).await;
}
#[tokio::test]
async fn reload_signal_resubscribes_with_the_new_set() {
let io = Arc::new(FakeIo::with_listings(vec![
vec![saved("channel_h", "channel-a", "[9]")],
vec![
saved("channel_h", "channel-a", "[9]"),
saved("owner_p", "owner-pk", "[24200]"),
],
]));
let (_tx, reload, cancel, handle) = spawn_sync(Arc::clone(&io));
wait_for("initial subscribe", || !io.applied().is_empty()).await;
reload.notify_one();
wait_for("resubscribe", || io.applied().len() >= 2).await;
assert_eq!(io.applied()[1].len(), 2);
stop(cancel, handle).await;
}
#[tokio::test(start_paused = true)]
async fn flushes_when_the_batch_size_is_reached() {
// Paused clock: the deadline can never fire, so a flush here is the size
// bound and nothing else. Without this the test passes on an off-by-one
// `is_full` — the deadline flushes the same 25 events 2s later and the
// assertion cannot tell the two apart.
let io = Arc::new(FakeIo::with_listings(vec![vec![saved(
"channel_h",
"channel-a",
"[9]",
)]]));
let (tx, _reload, cancel, handle) = spawn_sync(Arc::clone(&io));
wait_for("initial subscribe", || !io.applied().is_empty()).await;
let id = io.applied()[0][0].id.clone();
for _ in 0..(FLUSH_BATCH_SIZE - 1) {
tx.send(matched(&id)).await.unwrap();
}
// One short of the bound: nothing may flush.
wait_for("loop to drain the channel", || {
tx.capacity() == tx.max_capacity()
})
.await;
assert!(
io.archived().is_empty(),
"flushed before reaching the batch size"
);
tx.send(matched(&id)).await.unwrap();
wait_for("flush", || !io.archived().is_empty()).await;
// Exactly one batch of exactly FLUSH_BATCH_SIZE — a flush at the wrong
// boundary shows up here as a split or an oversized batch.
let archived = io.archived();
assert_eq!(archived.len(), 1);
assert_eq!(archived[0].len(), FLUSH_BATCH_SIZE);
assert!(archived[0].iter().all(|scope| scope == "channel-a"));
stop(cancel, handle).await;
}
#[tokio::test(start_paused = true)]
async fn flushes_a_partial_batch_after_the_deadline() {
let io = Arc::new(FakeIo::with_listings(vec![vec![saved(
"channel_h",
"channel-a",
"[9]",
)]]));
let (tx, _reload, cancel, handle) = spawn_sync(Arc::clone(&io));
wait_for("initial subscribe", || !io.applied().is_empty()).await;
let id = io.applied()[0][0].id.clone();
tx.send(matched(&id)).await.unwrap();
// Just under the deadline: still buffered.
tokio::time::sleep(FLUSH_DEADLINE - Duration::from_millis(1)).await;
assert!(io.archived().is_empty(), "flushed before the deadline");
tokio::time::sleep(Duration::from_millis(2)).await;
wait_for("flush", || !io.archived().is_empty()).await;
assert_eq!(io.archived()[0].len(), 1);
stop(cancel, handle).await;
}
#[tokio::test(start_paused = true)]
async fn a_trickle_cannot_postpone_the_deadline_indefinitely() {
// The deadline belongs to the OLDEST buffered event. An idle timer reset
// on each arrival would leave a steady trickle unflushed forever.
let io = Arc::new(FakeIo::with_listings(vec![vec![saved(
"channel_h",
"channel-a",
"[9]",
)]]));
let (tx, _reload, cancel, handle) = spawn_sync(Arc::clone(&io));
wait_for("initial subscribe", || !io.applied().is_empty()).await;
let id = io.applied()[0][0].id.clone();
for _ in 0..4 {
tx.send(matched(&id)).await.unwrap();
tokio::time::sleep(FLUSH_DEADLINE / 2).await;
}
wait_for("flush", || !io.archived().is_empty()).await;
assert!(!io.archived().is_empty());
stop(cancel, handle).await;
}
#[tokio::test]
async fn events_for_an_unknown_subscription_are_dropped() {
let io = Arc::new(FakeIo::with_listings(vec![vec![saved(
"channel_h",
"channel-a",
"[9]",
)]]));
let (tx, _reload, cancel, handle) = spawn_sync(Arc::clone(&io));
wait_for("initial subscribe", || !io.applied().is_empty()).await;
// An event from a subscription we already closed: no scope, no archive.
tx.send(matched("archive:channel_h:gone:[9]"))
.await
.unwrap();
cancel.cancel();
handle.await.unwrap();
assert!(
io.archived().is_empty(),
"archived an event with no known scope"
);
}
#[tokio::test]
async fn buffered_events_flush_on_shutdown() {
// Ephemeral kind 24200 cannot be re-fetched, so a buffered event dropped
// at teardown is lost permanently.
let io = Arc::new(FakeIo::with_listings(vec![vec![saved(
"owner_p", "owner-pk", "[24200]",
)]]));
let (tx, _reload, cancel, handle) = spawn_sync(Arc::clone(&io));
wait_for("initial subscribe", || !io.applied().is_empty()).await;
let id = io.applied()[0][0].id.clone();
tx.send(matched(&id)).await.unwrap();
// Wait until the loop has actually taken the event off the channel;
// cancelling first would test a race, not the shutdown flush.
wait_for("loop to drain the channel", || {
tx.capacity() == tx.max_capacity()
})
.await;
cancel.cancel();
handle.await.unwrap();
let archived = io.archived();
assert_eq!(archived.len(), 1, "shutdown did not flush the buffer");
assert_eq!(archived[0], vec!["owner-pk".to_string()]);
}
#[tokio::test]
async fn notifies_agent_metrics_only_when_the_backend_persisted_some() {
let io = Arc::new(FakeIo::with_listings(vec![vec![saved(
"owner_p", "owner-pk", "[44200]",
)]]));
*io.persisted_agent_metrics.lock().unwrap() = 2;
let (tx, _reload, cancel, handle) = spawn_sync(Arc::clone(&io));
wait_for("initial subscribe", || !io.applied().is_empty()).await;
let id = io.applied()[0][0].id.clone();
for _ in 0..FLUSH_BATCH_SIZE {
tx.send(matched(&id)).await.unwrap();
}
wait_for("flush", || !io.archived().is_empty()).await;
tokio::time::sleep(Duration::from_millis(10)).await;
assert_eq!(*io.metrics_notifications.lock().unwrap(), 1);
stop(cancel, handle).await;
}
#[tokio::test]
async fn does_not_notify_when_nothing_was_persisted() {
let io = Arc::new(FakeIo::with_listings(vec![vec![saved(
"owner_p", "owner-pk", "[44200]",
)]]));
// persisted_agent_metrics stays 0: a duplicate-only batch must not
// invalidate the renderer's usage queries.
let (tx, _reload, cancel, handle) = spawn_sync(Arc::clone(&io));
wait_for("initial subscribe", || !io.applied().is_empty()).await;
let id = io.applied()[0][0].id.clone();
for _ in 0..FLUSH_BATCH_SIZE {
tx.send(matched(&id)).await.unwrap();
}
wait_for("flush", || !io.archived().is_empty()).await;
tokio::time::sleep(Duration::from_millis(10)).await;
assert_eq!(*io.metrics_notifications.lock().unwrap(), 0);
stop(cancel, handle).await;
}
#[tokio::test]
async fn a_failed_archive_call_does_not_notify_or_stop_the_loop() {
let io = Arc::new(FakeIo::with_listings(vec![vec![saved(
"channel_h",
"channel-a",
"[9]",
)]]));
*io.archive_fails.lock().unwrap() = true;
let (tx, _reload, cancel, handle) = spawn_sync(Arc::clone(&io));
wait_for("initial subscribe", || !io.applied().is_empty()).await;
let id = io.applied()[0][0].id.clone();
for _ in 0..(FLUSH_BATCH_SIZE * 2) {
tx.send(matched(&id)).await.unwrap();
}
wait_for("second flush", || io.archived().len() >= 2).await;
assert_eq!(*io.metrics_notifications.lock().unwrap(), 0);
stop(cancel, handle).await;
}
#[tokio::test]
async fn a_failed_listing_leaves_the_previous_subscriptions_live() {
// A transient SQLite error must not silently stop archiving.
struct FailingList(Arc<FakeIo>, StdMutex<bool>);
impl ArchiveSyncIo for FailingList {
fn list_subscriptions(&self) -> BoxFuture<'_, Result<Vec<SaveSubscription>, String>> {
Box::pin(async move {
let mut failed = self.1.lock().unwrap();
if *failed {
return Err("db is busy".into());
}
*failed = true;
Ok(vec![saved("channel_h", "channel-a", "[9]")])
})
}
fn set_subscriptions(&self, subscriptions: Vec<Subscription>) -> BoxFuture<'_, ()> {
self.0.set_subscriptions(subscriptions)
}
fn archive(
&self,
candidates: Vec<ArchiveCandidate>,
) -> BoxFuture<'_, Result<ArchiveBatchResult, String>> {
self.0.archive(candidates)
}
fn notify_agent_metrics_changed(&self) {
self.0.notify_agent_metrics_changed();
}
}
let inner = Arc::new(FakeIo::default());
let io = Arc::new(FailingList(Arc::clone(&inner), StdMutex::new(false)));
let (tx, rx) = mpsc::channel(8);
let reload = Arc::new(Notify::new());
let cancel = CancellationToken::new();
let handle = {
let io = Arc::clone(&io);
let reload = Arc::clone(&reload);
let cancel = cancel.clone();
tokio::spawn(async move { run_sync(io.as_ref(), reload, rx, cancel).await })
};
wait_for("initial subscribe", || !inner.applied().is_empty()).await;
let id = inner.applied()[0][0].id.clone();
reload.notify_one();
tokio::time::sleep(Duration::from_millis(10)).await;
// The failed reload applied nothing, and the original scope still
// demultiplexes — so events keep being archived.
assert_eq!(inner.applied().len(), 1);
for _ in 0..FLUSH_BATCH_SIZE {
tx.send(matched(&id)).await.unwrap();
}
wait_for("flush", || !inner.archived().is_empty()).await;
stop(cancel, handle).await;
}
// ── Lifecycle ownership ──────────────────────────────────────────────────────
//
// The renderer fires `start_archive_sync` and `stop_archive_sync` without
// awaiting them, and Tauri commands may complete in any order. These tests
// drive `ArchiveSyncState` through the same `claim_start`/`claim_stop`/
// `install`/`stop` seam the commands use, applying the halves in a chosen
// order — which is the one thing the mounted-hook test cannot do, because its
// mock `invoke` resolves immediately and can only observe call order.
/// Runs the ownership half of `start_archive_sync` under `lease`, returning the
/// installed task's cancel token. `None` means the start did not take
/// ownership. Mirrors the command minus the relay session and spawned loop,
/// which the ordering invariant does not involve.
///
/// The token is the task's identity: two starts produce distinct tokens, so a
/// test can name WHICH task survived an interleaving rather than only that one
/// did. "Something is running" is satisfiable by the stale start's task.
async fn start_half(
state: &ArchiveSyncState,
mark: (u64, u64),
scope: (&str, &str),
) -> Option<CancellationToken> {
let cancel = CancellationToken::new();
state
.begin(
mark,
(scope.0.to_string(), scope.1.to_string()),
cancel.clone(),
Arc::new(Notify::new()),
)
.await
.then_some(cancel)
}
/// Runs `stop_archive_sync` under `mark`.
async fn stop_half(state: &ArchiveSyncState, mark: (u64, u64)) {
state.end(mark).await;
}
async fn is_running(state: &ArchiveSyncState) -> bool {
state.running.lock().await.is_some()
}
/// Whether the installed task is the one `cancel` belongs to.
///
/// `CancellationToken` is not `PartialEq`, so identity is checked through the
/// shared state the clones observe: cancelling the candidate must cancel the
/// installed task if and only if they are the same instance. The token is
/// consumed by the check, so callers assert identity last.
async fn running_is(state: &ArchiveSyncState, cancel: &CancellationToken) -> bool {
let installed = match state.running.lock().await.as_ref() {
Some(running) => running.cancel.clone(),
None => return false,
};
cancel.cancel();
installed.is_cancelled()
}
const SCOPE: (&str, &str) = ("owner-pubkey", "wss://relay.example");
/// Wren's schedule: `start2` reaches the backend before the delayed `start1`,
/// then the old effect's cleanup runs.
///
/// A backend-issued generation fails here — `start1` arrives last, so it mints
/// the newest token and hands it to the stalest caller, whose `stop` then
/// legitimately cancels the task the new effect depends on. The lease is
/// allocated in the renderer in effect order, so `start1` is stale on arrival.
#[tokio::test]
async fn a_start_that_arrives_after_a_newer_one_cannot_supersede_it() {
let state = ArchiveSyncState::default();
// Effect 2 wins the race to the backend.
let start2 = start_half(&state, (1, 2), SCOPE)
.await
.expect("start2 installs");
// Effect 1's delayed start lands second and must not take ownership.
assert!(
start_half(&state, (1, 1), SCOPE).await.is_none(),
"a start older than the newest lease must not install"
);
// Effect 1's cleanup, holding lease 1.
stop_half(&state, (1, 1)).await;
assert!(
!start2.is_cancelled(),
"start2's task must not have been cancelled by lease 1's stop"
);
// Identity, not survival: a lease mutant that keeps the WRONG task alive
// would satisfy "something is running", so name the instance.
assert!(
running_is(&state, &start2).await,
"the surviving task must be start2's instance — stale cleanup cancelled \
the newest start, the exact stranding this lease prevents"
);
}
/// The other half of the invariant: a start delayed past its own cleanup must
/// not resurrect sync after the renderer gate closed.
#[tokio::test]
async fn a_start_that_arrives_after_its_own_stop_cannot_resurrect_sync() {
let state = ArchiveSyncState::default();
stop_half(&state, (1, 3)).await;
assert!(
start_half(&state, (1, 3), SCOPE).await.is_none(),
"a start whose own stop already ran must not install"
);
assert!(
!is_running(&state).await,
"sync was resurrected after its owner stopped"
);
}
/// The ordinary sequence still works: each remount's start takes ownership and
/// its own cleanup stops it.
#[tokio::test]
async fn ordered_start_and_stop_still_take_effect() {
let state = ArchiveSyncState::default();
let first = start_half(&state, (1, 1), SCOPE)
.await
.expect("first start");
assert!(is_running(&state).await, "sync must be running after start");
stop_half(&state, (1, 1)).await;
assert!(!is_running(&state).await, "its own stop must take effect");
assert!(first.is_cancelled(), "the stopped task must be cancelled");
let second = start_half(&state, (1, 2), SCOPE)
.await
.expect("newer start");
assert!(
running_is(&state, &second).await,
"the newer start's own task must be the installed one"
);
stop_half(&state, (1, 2)).await;
assert!(!is_running(&state).await, "the newer stop must take effect");
}
/// A same-scope remount that reaches the backend in order is still a no-op at
/// the socket, so the lease does not undo the idempotence the port relies on.
#[tokio::test]
async fn a_same_scope_restart_does_not_churn_the_running_task() {
let state = ArchiveSyncState::default();
let first = start_half(&state, (1, 1), SCOPE)
.await
.expect("first start");
assert!(
start_half(&state, (1, 2), SCOPE).await.is_none(),
"a newer start for the same scope must not reinstall"
);
assert!(
!first.is_cancelled(),
"the original task must not be torn down"
);
assert!(
running_is(&state, &first).await,
"the original task must still be the installed one"
);
}
/// An identity or relay change must replace the task rather than leaving the
/// old scope's socket live.
#[tokio::test]
async fn a_scope_change_replaces_the_running_task() {
let state = ArchiveSyncState::default();
let first = start_half(&state, (1, 1), SCOPE)
.await
.expect("first start");
let second = start_half(&state, (1, 2), ("other-pubkey", SCOPE.1))
.await
.expect("a different scope must install");
assert!(
first.is_cancelled(),
"the replaced task must be cancelled, not leaked"
);
assert!(
running_is(&state, &second).await,
"the new scope's task must be the installed one"
);
}
/// A stop must hold its lease guard across the cancellation, not just across
/// the check.
///
/// The other lifecycle tests apply the two halves sequentially, so they cannot
/// see this: they pass against an `end` that releases `latest_lease` before
/// taking `running`. That version leaves a window — a stop clears its lease
/// check, yields, a newer start installs its task, and the resuming stop
/// cancels it. Stale cleanup strands the newest owner, which is the exact
/// failure the lease exists to prevent.
///
/// Rather than race it (unreliable either way), this observes the invariant
/// directly: hold `running` so a concurrent `end` must park after its lease
/// check, then ask whether `latest_lease` is still held. Held means the stop
/// and a competing start are mutually exclusive over the whole operation.
#[tokio::test]
async fn a_stop_holds_its_lease_guard_across_the_cancellation() {
let state = Arc::new(ArchiveSyncState::default());
start_half(&state, (1, 1), SCOPE)
.await
.expect("first start");
// Block the second half of `end` by owning the lock it must acquire.
let running_guard = state.running.lock().await;
let stopper = tokio::spawn({
let state = Arc::clone(&state);
async move { state.end((1, 2)).await }
});
// Let the stop run until it blocks on `running`.
tokio::task::yield_now().await;
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
// The parked stop is past its lease check. If it still holds the lease
// guard, no concurrent start can install a task for it to cancel.
let lease_held = state.latest.try_lock().is_err();
drop(running_guard);
stopper.await.expect("stop task");
assert!(
lease_held,
"a stop parked mid-cancellation released its lease guard: a newer start \
can install a task in that window, which this stop then cancels — \
stale cleanup stranding the newest owner"
);
}
// ── Realm epochs ─────────────────────────────────────────────────────────────
//
// A lease counter lives and dies with its JS realm, but this state lives for
// the whole Tauri process. A renderer reload (useReloadShortcut,
// RootErrorBoundary, useCommunityInit) restarts the counter at zero, so
// ordering on the lease alone makes every post-reload call look stale. The
// epoch is minted here so successive realms can be ordered by an authority
// that outlives them.
/// After a renderer reload, the new realm's first start must take ownership
/// even though its lease counter restarted below what the backend has seen.
///
/// This is the regression for the reload boundary: seed the backend as if a
/// realm had already run, then have a fresh realm announce and start from
/// lease 1. Ordering on the lease alone rejects it forever — and because the
/// RootErrorBoundary reload is the recovery path for a renderer crash, that
/// would make crash recovery the thing that permanently kills archive sync.
#[tokio::test]
async fn a_realm_that_reloaded_owns_sync_despite_restarting_its_lease() {
let state = ArchiveSyncState::default();
// Realm 1 ran and got as far as lease 2 (StrictMode alone reaches this).
let first_epoch = state.announce().await;
start_half(&state, (first_epoch, 1), SCOPE)
.await
.expect("realm 1 start");
stop_half(&state, (first_epoch, 2)).await;
// The realm is destroyed by reload; the JS lease counter restarts at 1.
let second_epoch = state.announce().await;
assert!(
second_epoch > first_epoch,
"each announcing realm must outrank the last"
);
let reloaded = start_half(&state, (second_epoch, 1), SCOPE)
.await
.expect("the first start after a reload must own sync");
assert!(
running_is(&state, &reloaded).await,
"the post-reload realm's task must be the installed one — ordering on \
the lease alone leaves sync permanently absent after any reload"
);
}
/// A call from a realm that has already been superseded must lose from the
/// moment the new realm ANNOUNCES — not merely once the new realm has managed
/// to land a lifecycle call of its own.
///
/// This is the stale-cleanup invariant replayed one level up: the dead realm's
/// in-flight cleanup arrives after the new realm has taken over, and a lease
/// comparison alone would let its larger counter win.
///
/// The old calls are sent BEFORE any epoch-2 lifecycle call deliberately.
/// That gap is reachable in production and can be arbitrarily long: the new
/// realm awaits its announcement, then waits on observer reconciliation before
/// it may issue a start at all. A version that mints the epoch without
/// publishing it passes any schedule where the new realm calls first, because
/// the new call — not the announcement — is what advanced the mark.
#[tokio::test]
async fn a_delayed_call_from_a_superseded_realm_cannot_supersede_the_new_one() {
let state = ArchiveSyncState::default();
// The old realm ran and owns the installed task.
let old_epoch = state.announce().await;
let stale = start_half(&state, (old_epoch, 1), SCOPE)
.await
.expect("the old realm installs");
// The new realm announces. It has issued no lifecycle call yet.
let new_epoch = state.announce().await;
// The dead realm's delayed start and cleanup, both with high leases, land
// in the gap between the new realm's announcement and its first call.
//
// The delayed start carries a DIFFERENT scope on purpose. Under `SCOPE` it
// would hit the same-scope remount no-op and return `None` whatever the
// mark said, so the assertion would hold against a backend that had stopped
// comparing marks entirely — a pass for a benign reason is not a pass.
assert!(
start_half(&state, (old_epoch, 99), ("stale-realm-pubkey", SCOPE.1))
.await
.is_none(),
"a superseded realm's start must not install, whatever its lease — \
announcing is what supersedes it, not the new realm's first call"
);
stop_half(&state, (old_epoch, 99)).await;
assert!(
!stale.is_cancelled(),
"a superseded realm's cleanup must not cancel a task it no longer owns"
);
// And the announcement must not have locked the new realm out of its own
// start: publishing a mark too high is blocker 3 rebuilt from the far side.
// A different scope so the start installs rather than taking the
// same-scope no-op path, which would report `None` for a benign reason and
// blur what this assertion is for.
let current = start_half(&state, (new_epoch, 1), ("other-pubkey", SCOPE.1))
.await
.expect("the announcing realm's own first start must still install");
assert!(
running_is(&state, &current).await,
"the surviving task must be the new realm's instance"
);
}
/// Epochs are handed out strictly increasing, so an announcement can never tie
/// with or fall behind one already given out.
#[tokio::test]
async fn announced_epochs_strictly_increase() {
let state = ArchiveSyncState::default();
let mut previous = 0;
for _ in 0..5 {
let epoch = state.announce().await;
assert!(
epoch > previous,
"epoch {epoch} did not outrank its predecessor {previous}"
);
previous = epoch;
}
}

View File

@@ -26,6 +26,7 @@ mod migration;
#[cfg(test)]
mod model_tests;
mod models;
mod native_relay_client;
mod native_websocket;
mod native_websocket_batch;
mod nostr_bind;
@@ -198,95 +199,10 @@ pub fn run() {
.plugin(tauri_plugin_dialog::init())
.plugin(tauri_plugin_process::init());
// The global-shortcut plugin is omitted from test builds: linking it into
// the lib-test binary makes it fail to load on Windows (STATUS_ENTRYPOINT_NOT_FOUND) before any test runs.
#[cfg(not(test))]
let builder = builder.plugin({
use tauri_plugin_global_shortcut::ShortcutState;
// Generation counter for the release delay task. Incremented on
// every press — a delayed release only fires if the generation
// hasn't changed (i.e. no new press happened during the delay).
// This prevents press→release→press within 200 ms from having
// the first release clobber the second press.
let ptt_press_gen = Arc::new(std::sync::atomic::AtomicU64::new(0));
tauri_plugin_global_shortcut::Builder::new()
.with_handler(move |app, _shortcut, event| {
let state = match app.try_state::<AppState>() {
Some(s) => s,
None => return,
};
// Only act if a huddle is active and mode is PTT.
let (is_ptt_mode, is_active) = match state.huddle_state.lock() {
Ok(hs) => (
hs.voice_input_mode == huddle::VoiceInputMode::PushToTalk,
matches!(
hs.phase,
huddle::HuddlePhase::Connected | huddle::HuddlePhase::Active
),
),
Err(_) => return,
};
if !is_ptt_mode || !is_active {
return;
}
match event.state {
ShortcutState::Pressed => {
// Bump generation — invalidates any pending release delay.
ptt_press_gen.fetch_add(1, std::sync::atomic::Ordering::Release);
if let Ok(hs) = state.huddle_state.lock() {
hs.ptt_active
.store(true, std::sync::atomic::Ordering::Release);
// Only cancel TTS if it's actually playing — avoids
// a stale cancel flag that drops the next queued message.
if hs.tts_active.load(std::sync::atomic::Ordering::Acquire) {
hs.tts_cancel
.store(true, std::sync::atomic::Ordering::Release);
}
}
// Emit ptt-state=true to the frontend.
// The React side plays the press audio cue on this event
// (Web Audio API via HuddleContext). Rust-side rodio audio
// was considered but rejected: the rodio OutputStream must
// outlive the handler and sharing it across the shortcut
// closure adds lifecycle complexity for marginal gain.
// The React implementation is sufficient and simpler.
let _ = app.emit("ptt-state", true);
}
ShortcutState::Released => {
// Capture generation at release time.
let gen_at_release =
ptt_press_gen.load(std::sync::atomic::Ordering::Acquire);
let gen_arc = Arc::clone(&ptt_press_gen);
let app_handle = app.clone();
// 200 ms release delay — captures the tail of the utterance.
// Only applies if no new press happened during the delay.
tauri::async_runtime::spawn(async move {
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
// Check generation — if it changed, a new press arrived.
if gen_arc.load(std::sync::atomic::Ordering::Acquire) != gen_at_release
{
return; // Superseded by a new press.
}
if let Some(state) = app_handle.try_state::<AppState>() {
if let Ok(hs) = state.huddle_state.lock() {
hs.ptt_active
.store(false, std::sync::atomic::Ordering::Release);
}
}
// Emit ptt-state=false — React plays the release audio cue.
let _ = app_handle.emit("ptt-state", false);
});
}
}
})
.build()
});
// The push-to-talk global-shortcut plugin lives in `ptt_shortcut`, next to
// the registration lifecycle it drives. Installing it is a no-op in test
// builds; see that module for why.
let builder = ptt_shortcut::install(builder);
// Register the updater only in configured release builds; omit it locally.
#[cfg(buzz_updater_enabled)]
@@ -312,6 +228,7 @@ pub fn run() {
.manage(BuilderlabLogin::default())
.manage(commands::pairing::PairingHandle::new())
.manage(terminal_runtime::TerminalSessions::default())
.manage(archive::sync::ArchiveSyncState::default())
.setup(move |app| {
let app_handle = app.handle().clone();
#[cfg(target_os = "macos")]
@@ -910,6 +827,9 @@ pub fn run() {
archive::index_observer_channel_id,
archive::read_unindexed_observer_rows,
archive::get_agent_usage_series,
archive::sync::announce_archive_sync_epoch,
archive::sync::start_archive_sync,
archive::sync::stop_archive_sync,
is_auto_update_supported,
set_window_vibrancy,
#[cfg(target_os = "macos")]

View File

@@ -0,0 +1,798 @@
//! Shared native relay session.
//!
//! Owns the authenticated relay socket for backend features that need live
//! subscriptions (archive sync today; persona catalog and catch-up next). One
//! session per (relay, pubkey) scope, multiplexing every subscription over a
//! single socket — a second socket per feature would multiply relay connection
//! slots and duplicate the NIP-42 handshake for no benefit.
//!
//! Built on `buzz-ws-client`, which owns the wire format and the NIP-42
//! handshake. That crate is request/response shaped (one caller, `next_event`
//! off a buffer); the session lifecycle lives here instead of being pushed down
//! into it, because `buzz-cli` and `buzz-test-client` consume that crate and do
//! not want subscription bookkeeping.
use std::{collections::HashMap, sync::Arc, time::Duration};
use buzz_ws_client_pkg::{NostrWsConnection, RelayMessage};
use nostr::{Event, Keys};
use tokio::{
sync::{mpsc, Mutex},
time::Instant,
};
use tokio_util::sync::CancellationToken;
/// Backoff floor for reconnect attempts.
const RECONNECT_BASE_DELAY: Duration = Duration::from_millis(500);
/// Backoff ceiling. Matches the renderer session's ceiling so a relay outage
/// produces one retry cadence across the app rather than two competing ones.
const RECONNECT_MAX_DELAY: Duration = Duration::from_secs(30);
/// How long a read may block before the loop re-checks cancellation. Not a
/// connection timeout: an idle relay is normal, so a lapsed read just loops.
const READ_TIMEOUT: Duration = Duration::from_secs(30);
/// Backoff floor for reopening a subscription the relay CLOSED. Matches
/// `RETRY_BASE_DELAY_MS` in `relayClosedRecovery.ts`.
const CLOSED_RETRY_BASE_DELAY: Duration = Duration::from_secs(1);
/// Backoff ceiling for reopening a CLOSED subscription. Matches
/// `RETRY_MAX_DELAY_MS` in `relayClosedRecovery.ts`.
const CLOSED_RETRY_MAX_DELAY: Duration = Duration::from_secs(30);
/// Delay for a `rate-limited:` CLOSED that carries no `retry in Ns` hint.
/// Matches `DEFAULT_RATE_LIMIT_SECONDS` on both sides of the client.
const CLOSED_RATE_LIMIT_DEFAULT: Duration = Duration::from_secs(10);
/// A live subscription request: a filter plus where its events go.
#[derive(Clone)]
pub(crate) struct Subscription {
/// Caller-stable key. Reused verbatim as the relay subscription id so a
/// resubscribe after reconnect replaces rather than duplicates.
pub(crate) id: String,
pub(crate) filter: serde_json::Value,
}
/// An event delivered to the session owner, tagged with the subscription that
/// matched it. Callers demultiplex on `subscription_id`.
pub(crate) struct MatchedEvent {
pub(crate) subscription_id: String,
pub(crate) event: Box<Event>,
}
/// Handle to a running session. Dropping it does not stop the session; call
/// [`RelaySession::shutdown`] so the socket closes deterministically.
pub(crate) struct RelaySession {
desired: Arc<Mutex<Vec<Subscription>>>,
wake: mpsc::Sender<()>,
cancel: CancellationToken,
}
impl RelaySession {
/// Replaces the desired subscription set and wakes the loop to reconcile.
///
/// Reconciliation is declarative rather than incremental: callers state
/// what they want and the loop diffs. An incremental add/remove API would
/// have to be replayed in order across a reconnect, which is exactly the
/// bug class this avoids.
///
/// It is also why no revision/generation guard is needed. Every reconcile
/// re-reads the current desired set, so a change that lands mid-pass is
/// picked up by the wake it queued rather than having to invalidate work
/// already in flight.
pub(crate) async fn set_subscriptions(&self, subscriptions: Vec<Subscription>) {
*self.desired.lock().await = subscriptions;
// A full channel already means "reconcile pending", so a failed send
// is success: the loop has not yet consumed the previous wake.
let _ = self.wake.try_send(());
}
pub(crate) fn shutdown(&self) {
self.cancel.cancel();
}
}
/// Starts a session against `relay_url` authenticated as `keys`.
///
/// Returns the handle plus the receiver for matched events. The session
/// reconnects on drop with exponential backoff and resubscribes the current
/// desired set — never a snapshot captured at connect time, so a subscription
/// change during an outage is honored by the reconnect that follows.
pub(crate) fn start(
relay_url: String,
keys: Keys,
auth_tag: Option<nostr::Tag>,
) -> (Arc<RelaySession>, mpsc::Receiver<MatchedEvent>) {
let (event_tx, event_rx) = mpsc::channel(256);
let (wake, wake_rx) = mpsc::channel(1);
let session = Arc::new(RelaySession {
desired: Arc::new(Mutex::new(Vec::new())),
wake,
cancel: CancellationToken::new(),
});
tauri::async_runtime::spawn(run_session(
relay_url,
keys,
auth_tag,
Arc::clone(&session),
wake_rx,
event_tx,
));
(session, event_rx)
}
async fn run_session(
relay_url: String,
keys: Keys,
auth_tag: Option<nostr::Tag>,
session: Arc<RelaySession>,
mut wake_rx: mpsc::Receiver<()>,
event_tx: mpsc::Sender<MatchedEvent>,
) {
let mut delay = RECONNECT_BASE_DELAY;
loop {
if session.cancel.is_cancelled() {
return;
}
match NostrWsConnection::connect_authenticated(&relay_url, &keys, auth_tag.as_ref()).await {
Ok(conn) => {
// A connection that authenticated is healthy regardless of how
// long it then lived, so backoff resets here rather than on
// clean exit — a socket that drops after one event must not
// inherit the previous failure's delay.
delay = RECONNECT_BASE_DELAY;
run_connection(conn, &session, &mut wake_rx, &event_tx).await;
}
Err(error) => {
eprintln!("buzz-desktop: native_relay_client: connect failed: {error}");
}
}
if session.cancel.is_cancelled() {
return;
}
tokio::select! {
_ = session.cancel.cancelled() => return,
_ = tokio::time::sleep(delay) => {}
}
delay = (delay * 2).min(RECONNECT_MAX_DELAY);
}
}
/// Drives one connected socket until it drops or the session is cancelled.
async fn run_connection(
mut conn: NostrWsConnection,
session: &RelaySession,
wake_rx: &mut mpsc::Receiver<()>,
event_tx: &mpsc::Sender<MatchedEvent>,
) {
// Subscription ids currently open ON THIS SOCKET. Deliberately local: a new
// socket has none, so reconnect resubscribes the full desired set without
// any explicit "resubscribe" path that could drift from the normal one.
let mut open: HashMap<String, serde_json::Value> = HashMap::new();
// Reopen schedule for ids the relay CLOSED, keyed the same way and equally
// local — for the same reason and one more. Backoff state cannot live in
// `desired`: that set is reloaded from SQLite by the archive task, so a
// subscription deleted there is re-added by the next reload. The JS port
// could delete from its subscription map because that map WAS the desired
// set; here the two are separate, and only this one is per-socket.
let mut retries: HashMap<String, ClosedRetry> = HashMap::new();
if !reconcile(&mut conn, session, &mut open, &retries).await {
return;
}
loop {
// Earliest pending reopen, or `None` when nothing is scheduled. The arm
// below is disabled in that case rather than sleeping on a far-future
// instant, so an idle connection never wakes on this branch.
let retry_at = retries.values().filter_map(|retry| retry.due_at).min();
tokio::select! {
_ = session.cancel.cancelled() => {
let _ = conn.disconnect().await;
return;
}
Some(()) = wake_rx.recv() => {
if !reconcile(&mut conn, session, &mut open, &retries).await {
return;
}
}
// The edge that makes a CLOSED recoverable. Without it, nothing
// re-enters `reconcile` unless the desired set changes again, and
// for a stable set that means the subscription is dead for the life
// of the socket.
_ = tokio::time::sleep_until(retry_at.unwrap_or_else(Instant::now)),
if retry_at.is_some() =>
{
for retry in retries.values_mut() {
if retry.due_at.is_some_and(|due| due <= Instant::now()) {
retry.due_at = None;
}
}
if !reconcile(&mut conn, session, &mut open, &retries).await {
return;
}
}
message = conn.next_event(READ_TIMEOUT) => {
match message {
Ok(RelayMessage::Event { subscription_id, event }) => {
// Only forward events for a subscription we still want.
// A CLOSE races in flight with events already queued at
// the relay, so this is the last line of defense
// against delivering out-of-scope events after a change.
if !open.contains_key(&subscription_id) {
continue;
}
// Delivery proves the subscription is healthy, so any
// accumulated backoff for it is stale. Mirrors the JS
// port's per-event `closedRetryAttempt = 0`.
retries.remove(&subscription_id);
if event_tx
.send(MatchedEvent { subscription_id, event })
.await
.is_err()
{
// Receiver gone: nobody is consuming this session.
let _ = conn.disconnect().await;
return;
}
}
Ok(RelayMessage::Closed { subscription_id, message }) => {
// The relay dropped it; forget it so a reopen re-sends
// REQ rather than assuming it is still live.
open.remove(&subscription_id);
let retry = retries.entry(subscription_id.clone()).or_default();
retry.schedule(&message);
eprintln!(
"buzz-desktop: native_relay_client: relay closed {subscription_id}: {message}"
);
}
Ok(RelayMessage::Eose { subscription_id }) => {
// The relay served this subscription, so whatever
// caused an earlier CLOSED has cleared. Same reset the
// JS port performs in `handleSubscriptionEose`, and it
// is what keeps an intermittent relay from ratcheting
// its way to the 30s ceiling and staying there.
retries.remove(&subscription_id);
}
Ok(_) => {}
Err(error) => {
if !is_read_timeout(&error) {
eprintln!("buzz-desktop: native_relay_client: read failed: {error}");
return;
}
}
}
}
}
}
}
/// Brings the socket's open subscriptions in line with the desired set.
///
/// Returns false when the socket failed and the caller should reconnect.
async fn reconcile(
conn: &mut NostrWsConnection,
session: &RelaySession,
open: &mut HashMap<String, serde_json::Value>,
retries: &HashMap<String, ClosedRetry>,
) -> bool {
let desired = session.desired.lock().await.clone();
for id in open.keys().cloned().collect::<Vec<_>>() {
if desired.iter().any(|s| s.id == id) {
continue;
}
if conn
.send_raw(&serde_json::json!(["CLOSE", id]))
.await
.is_err()
{
return false;
}
open.remove(&id);
}
for sub in desired {
// A filter change under the same id must reopen, not be skipped: the
// relay replaces a subscription by id, so re-sending REQ is the update.
if open.get(&sub.id) == Some(&sub.filter) {
continue;
}
// Held back by a CLOSED: either waiting out its backoff, or terminal
// and never to be retried on this socket. Both are `is_blocked`, which
// is what keeps a relay that rejects on policy from being re-asked at
// the speed of the event loop.
if retries.get(&sub.id).is_some_and(ClosedRetry::is_blocked) {
continue;
}
if conn
.send_raw(&serde_json::json!(["REQ", sub.id, sub.filter]))
.await
.is_err()
{
return false;
}
open.insert(sub.id, sub.filter);
}
true
}
/// Reopen schedule for one subscription the relay CLOSED.
#[derive(Default)]
struct ClosedRetry {
/// When the reopen is due. `None` means "not waiting": either the delay has
/// elapsed and reconcile may re-send, or `terminal` latched.
due_at: Option<Instant>,
/// Consecutive CLOSEDs, driving the exponential delay. Reset by a delivered
/// event or EOSE, both of which drop the whole entry.
attempts: u32,
/// The relay rejected this filter for a reason retrying cannot change.
terminal: bool,
}
impl ClosedRetry {
/// True while reconcile must leave this subscription closed.
fn is_blocked(&self) -> bool {
self.terminal || self.due_at.is_some_and(|due| due > Instant::now())
}
/// Records a CLOSED and schedules the reopen its class calls for.
fn schedule(&mut self, message: &str) {
match classify_closed(message) {
// Auth, access, or filter errors will fail identically until
// something outside this socket changes, so stop asking. Scoped to
// this socket by construction: the state lives in `run_connection`,
// so a reconnect retries once through the normal path. That is
// deliberate — relay policy and our own auth can change across a
// reconnect, and one REQ per reconnect is bounded.
ClosedClass::Terminal => {
self.terminal = true;
self.due_at = None;
}
ClosedClass::RateLimited => {
// Arm the process-wide gate so the HTTP bridge backs off too,
// rather than keeping a second private notion of the same
// relay's back-pressure.
let hint = parse_retry_in_seconds(message);
crate::relay_admission::activate_rate_limit(hint);
let hinted = hint
.map(Duration::from_secs)
.unwrap_or(CLOSED_RATE_LIMIT_DEFAULT);
// The longer of the two: a short hint must not undercut a
// backoff already grown by repeated rejections.
self.due_at = Some(Instant::now() + self.backoff().max(hinted));
self.attempts = self.attempts.saturating_add(1);
}
ClosedClass::Retryable => {
self.due_at = Some(Instant::now() + self.backoff());
self.attempts = self.attempts.saturating_add(1);
}
}
}
/// Exponential delay for the current attempt, capped. The shift is bounded
/// before it is taken, so a long-lived rejection cannot overflow its way
/// back down to a short delay.
fn backoff(&self) -> Duration {
CLOSED_RETRY_BASE_DELAY
.saturating_mul(1_u32 << self.attempts.min(16))
.min(CLOSED_RETRY_MAX_DELAY)
}
}
/// How a CLOSED message should be handled.
///
/// Ported from `classifyRelayClosed` in `relayClosedPolicy.ts`; the prefixes are
/// the relay's own machine-readable NIP-01 classes and must stay in step with
/// that file.
#[derive(Debug, PartialEq, Eq)]
enum ClosedClass {
Retryable,
RateLimited,
Terminal,
}
fn classify_closed(message: &str) -> ClosedClass {
let normalized = message.trim().to_ascii_lowercase();
if normalized.starts_with("rate-limited:") {
return ClosedClass::RateLimited;
}
// `auth-required:` is deliberately absent, i.e. retryable: it occurs
// transiently when a REQ races the AUTH handshake after a reconnect, and
// the backoff reopen re-sends once authenticated. A session that is
// genuinely unauthenticated fails at `connect_authenticated` instead, so
// this cannot loop forever.
if [
"restricted:",
"blocked:",
"invalid:",
"pow:",
"duplicate:",
"unsupported:",
"error: mixed search",
"error: too many subscriptions",
]
.iter()
.any(|prefix| normalized.starts_with(prefix))
{
return ClosedClass::Terminal;
}
ClosedClass::Retryable
}
/// Parses the relay's canonical `retry in Ns` hint. Same format the HTTP bridge
/// parses in `relay::extract_retry_in_hint`.
fn parse_retry_in_seconds(message: &str) -> Option<u64> {
let after = &message[message.find("retry in ")? + "retry in ".len()..];
after
.chars()
.take_while(char::is_ascii_digit)
.collect::<String>()
.parse()
.ok()
}
/// A lapsed read is an idle relay, not a failure. Distinguished by variant
/// rather than by message text so a reworded error cannot turn every idle
/// period into a reconnect storm.
fn is_read_timeout(error: &buzz_ws_client_pkg::WsClientError) -> bool {
matches!(error, buzz_ws_client_pkg::WsClientError::Timeout)
}
#[cfg(test)]
mod closed_recovery_tests {
use super::*;
use futures_util::{SinkExt, StreamExt};
use tokio_tungstenite::tungstenite::protocol::Message;
/// Minimal relay that completes the NIP-42 handshake, records every REQ,
/// and sends a CLOSED only when the test asks it to.
///
/// A real socket rather than a fake `NostrWsConnection`, because the bug
/// this covers lives in the lifecycle between frames — the loop's only
/// reconcile triggers — and a fake that hands the loop a `Closed` value
/// cannot show that a REQ went back out over the wire afterwards. Same
/// `accept_async` stub shape as `native_websocket.rs`'s live-TCP tests.
///
/// CLOSED is test-driven rather than a scripted reply to the first REQ so
/// the test can wait for the session to go quiet first. `set_subscriptions`
/// queues a wake that may still be pending when an immediate CLOSED lands,
/// and that wake reopens the subscription on its own — which made the first
/// version of this test pass against the unfixed code.
async fn stub_relay() -> (
String,
mpsc::Receiver<String>,
mpsc::Sender<(String, String)>,
) {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("bind stub relay");
let address = listener.local_addr().expect("stub relay address");
let (req_tx, req_rx) = mpsc::channel(16);
let (closed_tx, mut closed_rx) = mpsc::channel::<(String, String)>(4);
tokio::spawn(async move {
let (stream, _) = listener.accept().await.expect("accept");
let mut socket = tokio_tungstenite::accept_async(stream)
.await
.expect("websocket handshake");
socket
.send(Message::Text(r#"["AUTH","stub-challenge"]"#.into()))
.await
.expect("send challenge");
loop {
tokio::select! {
incoming = socket.next() => {
let Some(Ok(Message::Text(text))) = incoming else { return };
let Ok(frame) = serde_json::from_str::<serde_json::Value>(&text) else {
continue;
};
match frame[0].as_str() {
Some("AUTH") => {
let id = frame[1]["id"].as_str().unwrap_or_default();
socket
.send(Message::Text(
serde_json::json!(["OK", id, true, ""]).to_string().into(),
))
.await
.expect("send auth ok");
}
Some("REQ") => {
let id = frame[1].as_str().unwrap_or_default().to_string();
if req_tx.send(id).await.is_err() {
return;
}
}
_ => {}
}
}
Some((id, message)) = closed_rx.recv() => {
socket
.send(Message::Text(
serde_json::json!(["CLOSED", id, message]).to_string().into(),
))
.await
.expect("send closed");
}
}
}
});
(format!("ws://{address}"), req_rx, closed_tx)
}
fn probe_subscription() -> Subscription {
Subscription {
id: "archive:probe".to_string(),
filter: serde_json::json!({ "kinds": [1], "limit": 0 }),
}
}
async fn next_req(reqs: &mut mpsc::Receiver<String>, label: &str) -> String {
tokio::time::timeout(Duration::from_secs(10), reqs.recv())
.await
.unwrap_or_else(|_| panic!("timed out waiting for {label}"))
.unwrap_or_else(|| panic!("stub relay closed before {label}"))
}
/// Waits out the wake `set_subscriptions` queued, so a CLOSED sent after
/// this cannot be reopened by anything but the CLOSED path itself.
///
/// A pending wake is harmless while the subscription is still open — that
/// reconcile is a no-op — so draining it before the CLOSED is what makes
/// the assertion below attributable.
async fn settle() {
tokio::time::sleep(Duration::from_millis(500)).await;
}
/// The blocker: a CLOSED with the desired set never changing again must
/// still reopen the subscription.
///
/// Before the fix the loop removed the id from `open` and waited on a wake
/// that only `set_subscriptions` can produce, so a stable desired set left
/// the subscription dead for the life of the socket — silent permanent
/// loss for ephemeral kind 24200.
#[tokio::test]
async fn a_closed_subscription_reopens_without_a_desired_set_change() {
let (relay_url, mut reqs, closed) = stub_relay().await;
let (session, _events) = start(relay_url, Keys::generate(), None);
session.set_subscriptions(vec![probe_subscription()]).await;
assert_eq!(
next_req(&mut reqs, "the initial REQ").await,
"archive:probe"
);
settle().await;
// Retryable class, sent once: the reopen is answered normally, so a
// failure here means "never retried" rather than "retried into another
// rejection".
closed
.send(("archive:probe".into(), "error: temporary".into()))
.await
.expect("stub relay accepts the closed command");
// No `set_subscriptions` between the two REQs: the reopen must come
// from the CLOSED itself, which is exactly the edge that was missing.
assert_eq!(
next_req(&mut reqs, "the reopened REQ").await,
"archive:probe"
);
session.shutdown();
}
/// A relay that rejects on policy must not be re-asked in a tight loop.
#[tokio::test]
async fn a_terminal_closed_is_not_retried_on_the_same_socket() {
let (relay_url, mut reqs, closed) = stub_relay().await;
let (session, _events) = start(relay_url, Keys::generate(), None);
session.set_subscriptions(vec![probe_subscription()]).await;
assert_eq!(
next_req(&mut reqs, "the initial REQ").await,
"archive:probe"
);
settle().await;
closed
.send(("archive:probe".into(), "restricted: not authorized".into()))
.await
.expect("stub relay accepts the closed command");
// Long enough that a retryable class (1s base) would have reopened
// several times, so this asserts suppression rather than just slowness.
let retried = tokio::time::timeout(Duration::from_secs(5), reqs.recv()).await;
assert!(
retried.is_err(),
"a terminal CLOSED must not be retried on this socket, got {retried:?}"
);
session.shutdown();
}
#[test]
fn closed_messages_classify_like_the_renderer_policy() {
assert_eq!(
classify_closed("rate-limited: quota exceeded; retry in 4s"),
ClosedClass::RateLimited
);
assert_eq!(
classify_closed("restricted: not authorized"),
ClosedClass::Terminal
);
assert_eq!(
classify_closed("error: too many subscriptions"),
ClosedClass::Terminal
);
// Transient AUTH race, not a permanent rejection — the one prefix that
// looks terminal and deliberately is not.
assert_eq!(
classify_closed("auth-required: we can't serve unauthenticated"),
ClosedClass::Retryable
);
assert_eq!(classify_closed(""), ClosedClass::Retryable);
// Case and padding come from the relay, not from us.
assert_eq!(
classify_closed(" RESTRICTED: nope "),
ClosedClass::Terminal
);
}
#[test]
fn retry_delay_grows_and_stops_at_the_ceiling() {
let mut retry = ClosedRetry::default();
assert_eq!(retry.backoff(), CLOSED_RETRY_BASE_DELAY);
retry.schedule("error: temporary");
assert_eq!(retry.backoff(), CLOSED_RETRY_BASE_DELAY * 2);
for _ in 0..40 {
retry.schedule("error: temporary");
}
assert_eq!(
retry.backoff(),
CLOSED_RETRY_MAX_DELAY,
"backoff must saturate at the ceiling rather than wrapping"
);
}
#[test]
fn a_rate_limited_closed_waits_at_least_the_relay_hint() {
let mut retry = ClosedRetry::default();
retry.schedule("rate-limited: quota exceeded; retry in 12s");
let due = retry.due_at.expect("rate-limited must schedule a reopen");
// The hint dominates the 1s first backoff, so this asserts the hint was
// honored rather than that anything at all was scheduled.
assert!(
due >= Instant::now() + Duration::from_secs(11),
"a 12s hint must not be undercut by the base backoff"
);
crate::relay_admission::reset_rate_limit_gate();
}
#[test]
fn a_hintless_rate_limited_closed_uses_the_shared_default() {
let mut retry = ClosedRetry::default();
retry.schedule("rate-limited: quota exceeded");
let due = retry.due_at.expect("rate-limited must schedule a reopen");
assert!(
due >= Instant::now() + CLOSED_RATE_LIMIT_DEFAULT - Duration::from_secs(1),
"a hintless rate-limit must fall back to the shared default window"
);
crate::relay_admission::reset_rate_limit_gate();
}
#[test]
fn retry_hints_parse_the_relays_canonical_format() {
assert_eq!(
parse_retry_in_seconds("rate-limited: quota exceeded; retry in 4s"),
Some(4)
);
assert_eq!(parse_retry_in_seconds("rate-limited: quota exceeded"), None);
assert_eq!(parse_retry_in_seconds("retry in s"), None);
}
}
#[cfg(test)]
mod relay_backed_tests {
use super::*;
use nostr::{EventBuilder, Tag};
/// Relay-backed proof that the session's wire shape is one a real relay
/// accepts and answers.
///
/// Every other test in this commit drives `run_sync` through a fake
/// [`crate::archive::sync::ArchiveSyncIo`], which is the right default:
/// batching and demultiplexing are the logic worth pinning, and they must
/// not need a socket. But a fake cannot fail the one way this layer
/// actually can — by sending a REQ the relay rejects, or by filtering on a
/// tag key that matches nothing. The JS manager's filters were validated by
/// years of production traffic; this port's have been validated by my
/// reading of that code, which is exactly the claim a real relay can check
/// and I cannot.
///
/// `#[ignore]`d because it needs a relay on `BUZZ_TEST_RELAY_URL`. Run:
///
/// ```text
/// ./scripts/start-isolated-test-relay.sh # ws://localhost:3030
/// BUZZ_TEST_RELAY_URL=ws://localhost:3030 \
/// cargo test -p buzz-desktop -- --ignored archive_sync_session
/// ```
#[tokio::test]
#[ignore = "requires a local relay (set BUZZ_TEST_RELAY_URL)"]
async fn archive_sync_session_receives_live_events_from_a_real_relay() {
let Ok(relay_url) = std::env::var("BUZZ_TEST_RELAY_URL") else {
panic!("set BUZZ_TEST_RELAY_URL to a running relay");
};
let owner = Keys::generate();
let author = Keys::generate();
let owner_pk = owner.public_key();
// Kind 1 rather than the archive's own kind 24200. Publishing a real
// observer frame requires a registered agent-owner binding in the
// relay's database — a relay ACL concern that says nothing about this
// layer. What this test can prove, and what no fake can, is the wire
// shape: that the `#p` tag key and the `limit: 0` live tail produce a
// REQ a real relay accepts and answers. Scope demultiplexing on the
// archive side is covered in `archive/sync_tests.rs`.
let (session, mut events) = start(relay_url.clone(), owner.clone(), None);
session
.set_subscriptions(vec![Subscription {
id: "archive:owner_p:test".to_string(),
filter: serde_json::json!({
"kinds": [1],
"limit": 0,
"#p": [owner_pk.to_hex()],
}),
}])
.await;
// The subscription must be live at the relay before the event is
// published. A `limit: 0` filter is a live tail: it replays nothing,
// so anything published into a not-yet-open subscription is missed.
// That is the same ordering hazard the renderer start gate exists to
// prevent for the ephemeral archive kind.
tokio::time::sleep(Duration::from_secs(1)).await;
let mut publisher = NostrWsConnection::connect_authenticated(&relay_url, &author, None)
.await
.expect("publisher connect");
let frame = EventBuilder::text_note("archive-sync-probe")
.tag(Tag::public_key(owner_pk))
.sign_with_keys(&author)
.expect("sign event");
let frame_id = frame.id.to_hex();
let ok = publisher.send_event(frame).await.expect("publish frame");
assert!(
ok.accepted,
"relay rejected the observer frame, so a delivery timeout below would \
blame the subscription for a publish failure: {}",
ok.message
);
let received = tokio::time::timeout(Duration::from_secs(10), events.recv())
.await
.expect("timed out waiting for the relay to deliver the frame")
.expect("session channel closed");
assert_eq!(
received.subscription_id, "archive:owner_p:test",
"delivered event must carry the subscription id the loop demultiplexes on"
);
assert_eq!(
received.event.id.to_hex(),
frame_id,
"must deliver the published frame"
);
session.shutdown();
}
}

View File

@@ -8,6 +8,111 @@
use crate::huddle::HuddleState;
#[cfg(not(test))]
use crate::huddle::{HuddlePhase, VoiceInputMode};
use tauri::{Builder, Runtime};
/// Install the global-shortcut plugin and its push-to-talk key handler.
///
/// No-op in test builds: linking the plugin into the lib-test binary makes it
/// fail to load on Windows (STATUS_ENTRYPOINT_NOT_FOUND) before any test runs.
/// `sync_registration` is stubbed out under the same cfg for the same reason.
#[cfg(test)]
pub fn install<R: Runtime>(builder: Builder<R>) -> Builder<R> {
builder
}
/// Install the global-shortcut plugin and its push-to-talk key handler.
///
/// Registration itself is driven by huddle state through [`sync_registration`];
/// this only installs the plugin the handler runs on.
#[cfg(not(test))]
pub fn install<R: Runtime>(builder: Builder<R>) -> Builder<R> {
use crate::app_state::AppState;
use std::sync::Arc;
use tauri::{Emitter, Manager};
use tauri_plugin_global_shortcut::ShortcutState;
// Generation counter for the release delay task. Incremented on
// every press — a delayed release only fires if the generation
// hasn't changed (i.e. no new press happened during the delay).
// This prevents press→release→press within 200 ms from having
// the first release clobber the second press.
let ptt_press_gen = Arc::new(std::sync::atomic::AtomicU64::new(0));
builder.plugin(
tauri_plugin_global_shortcut::Builder::new()
.with_handler(move |app, _shortcut, event| {
let state = match app.try_state::<AppState>() {
Some(s) => s,
None => return,
};
// Only act if a huddle is active and mode is PTT.
let (is_ptt_mode, is_active) = match state.huddle_state.lock() {
Ok(hs) => (
hs.voice_input_mode == VoiceInputMode::PushToTalk,
matches!(hs.phase, HuddlePhase::Connected | HuddlePhase::Active),
),
Err(_) => return,
};
if !is_ptt_mode || !is_active {
return;
}
match event.state {
ShortcutState::Pressed => {
// Bump generation — invalidates any pending release delay.
ptt_press_gen.fetch_add(1, std::sync::atomic::Ordering::Release);
if let Ok(hs) = state.huddle_state.lock() {
hs.ptt_active
.store(true, std::sync::atomic::Ordering::Release);
// Only cancel TTS if it's actually playing — avoids
// a stale cancel flag that drops the next queued message.
if hs.tts_active.load(std::sync::atomic::Ordering::Acquire) {
hs.tts_cancel
.store(true, std::sync::atomic::Ordering::Release);
}
}
// Emit ptt-state=true to the frontend.
// The React side plays the press audio cue on this event
// (Web Audio API via HuddleContext). Rust-side rodio audio
// was considered but rejected: the rodio OutputStream must
// outlive the handler and sharing it across the shortcut
// closure adds lifecycle complexity for marginal gain.
// The React implementation is sufficient and simpler.
let _ = app.emit("ptt-state", true);
}
ShortcutState::Released => {
// Capture generation at release time.
let gen_at_release =
ptt_press_gen.load(std::sync::atomic::Ordering::Acquire);
let gen_arc = Arc::clone(&ptt_press_gen);
let app_handle = app.clone();
// 200 ms release delay — captures the tail of the utterance.
// Only applies if no new press happened during the delay.
tauri::async_runtime::spawn(async move {
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
// Check generation — if it changed, a new press arrived.
if gen_arc.load(std::sync::atomic::Ordering::Acquire) != gen_at_release
{
return; // Superseded by a new press.
}
if let Some(state) = app_handle.try_state::<AppState>() {
if let Ok(hs) = state.huddle_state.lock() {
hs.ptt_active
.store(false, std::sync::atomic::Ordering::Release);
}
}
// Emit ptt-state=false — React plays the release audio cue.
let _ = app_handle.emit("ptt-state", false);
});
}
}
})
.build(),
)
}
/// Whether the PTT shortcut should currently be reserved with the OS.
#[cfg(not(test))]

View File

@@ -57,7 +57,8 @@ import {
useUserStatusSubscription,
} from "@/features/user-status/hooks";
import { useCommunityEmojiLiveUpdates } from "@/features/custom-emoji/hooks";
import { useArchiveSync } from "@/features/local-archive/archiveSyncManager";
import { useArchiveSync } from "@/features/local-archive/useArchiveSync";
import { useArchiveAgentMetricsBridge } from "@/features/local-archive/useArchiveAgentMetricsBridge";
import { useObserverArchiveReconciliation } from "@/features/local-archive/useObserverArchiveSeed";
import { useAgentMetricArchiveSeed } from "@/features/local-archive/useAgentMetricArchiveSeed";
import { useProfileQuery } from "@/features/profile/hooks";
@@ -205,6 +206,9 @@ export function AppShell() {
// useArchiveSync must wait for reconciliation, or listeners could open
// before kind 24200 is guaranteed present in the subscription.
useArchiveSync(observerReconciled);
// The archive batch now persists in Rust, so the agent-metrics invalidation
// signal arrives as a Tauri event rather than an in-process call.
useArchiveAgentMetricsBridge();
// Kind 44200 is relay-persisted (durable) and stays deferred: missed
// startup frames can be replayed, so there's no ordering constraint.
const deferredPubkey = startupReady ? identityQuery.data?.pubkey : undefined;

File diff suppressed because it is too large Load Diff

View File

@@ -1,345 +0,0 @@
import { relayClient as defaultRelayClient } from "@/shared/api/relayClient";
import type { RelaySubscriptionFilter } from "@/shared/api/relayClientShared";
import type { RelayEvent } from "@/shared/api/types";
import {
archiveEvents as defaultArchiveEvents,
listSaveSubscriptions as defaultListSaveSubscriptions,
notifyAgentMetricsChanged,
onSubscriptionChange as defaultOnSubscriptionChange,
type ArchiveBatchResult,
type SaveSubscription,
type ScopeType,
} from "@/shared/api/tauriArchive";
// ── Constants ─────────────────────────────────────────────────────────────────
const FLUSH_BATCH_SIZE = 25;
const FLUSH_IDLE_MS = 2_000;
// ── Types ─────────────────────────────────────────────────────────────────────
/** Dependency injection interface — production uses module singletons; tests inject fakes. */
export interface ArchiveSyncDeps {
relayClient: {
subscribeLive: (
filter: RelaySubscriptionFilter,
onEvent: (event: RelayEvent) => void,
) => Promise<() => Promise<void>>;
};
listSaveSubscriptions: () => Promise<SaveSubscription[]>;
archiveEvents: (
candidates: Array<{
rawEventJson: string;
matchedScope: { scopeType: ScopeType; scopeValue: string };
}>,
) => Promise<ArchiveBatchResult>;
onSubscriptionChange: (listener: () => void) => () => void;
flushBatchSize?: number;
flushIdleMs?: number;
}
// ── Helpers ───────────────────────────────────────────────────────────────────
function buildFilter(sub: SaveSubscription): RelaySubscriptionFilter {
const base = { kinds: sub.kinds, limit: 0 } as const;
switch (sub.scopeType) {
case "channel_h":
return { ...base, "#h": [sub.scopeValue] };
case "owner_p":
return { ...base, "#p": [sub.scopeValue] };
case "referenced_e":
return { ...base, "#e": [sub.scopeValue] };
}
}
/** Stable key encoding scope + kinds — ensures kinds changes trigger resubscribe. */
function subKey(
scopeType: ScopeType,
scopeValue: string,
kinds: number[],
): string {
const sortedKinds = [...kinds].sort((a, b) => a - b).join(",");
return `${scopeType}:${scopeValue}:${sortedKinds}`;
}
/** Scope-only key used to find and tear down a stale sub when kinds change. */
function scopeKey(scopeType: ScopeType, scopeValue: string): string {
return `${scopeType}:${scopeValue}`;
}
// ── ArchiveSyncManager ────────────────────────────────────────────────────────
/**
* Always-on manager that opens one live relay subscription per saved archive
* config and forwards matched events to `archive_events` in debounced batches.
*
* Lifecycle: created once at app-shell mount (see `useArchiveSync`), destroyed
* on community switch. Resubscribes automatically when subscriptions change
* via the module-level notifier in `tauriArchive.ts`.
*
* Accepts optional `deps` for testing — production callers pass nothing.
*/
export class ArchiveSyncManager {
private readonly deps: Required<
Omit<ArchiveSyncDeps, "flushBatchSize" | "flushIdleMs">
>;
private readonly flushBatchSize: number;
private readonly flushIdleMs: number;
// full subKey (scope+kinds) → unsub
private active = new Map<string, () => Promise<void>>();
// Single-flight reload state — exactly one doResubscribe body runs at a time.
// Any reload request arriving while one is running sets reloadPending so that
// the loop runs one additional full pass after the current one finishes.
// This structural guarantee makes concurrent list/subscribe awaits impossible,
// eliminating all interleaving defects without per-boundary guards.
private reloading = false;
private reloadPending = false;
private buffer: Array<{
rawEventJson: string;
matchedScope: { scopeType: ScopeType; scopeValue: string };
}> = [];
private flushTimer: ReturnType<typeof setTimeout> | null = null;
private destroyed = false;
private offSubscriptionChange: (() => void) | null = null;
constructor(deps?: ArchiveSyncDeps) {
this.deps = {
relayClient: deps?.relayClient ?? defaultRelayClient,
listSaveSubscriptions:
deps?.listSaveSubscriptions ?? defaultListSaveSubscriptions,
archiveEvents: deps?.archiveEvents ?? defaultArchiveEvents,
onSubscriptionChange:
deps?.onSubscriptionChange ?? defaultOnSubscriptionChange,
};
this.flushBatchSize = deps?.flushBatchSize ?? FLUSH_BATCH_SIZE;
this.flushIdleMs = deps?.flushIdleMs ?? FLUSH_IDLE_MS;
}
async start(): Promise<void> {
// Register the change listener before the initial load so that any
// subscription change arriving while the first pass is running sets
// reloadPending and gets picked up by the coalescing loop.
this.offSubscriptionChange = this.deps.onSubscriptionChange(() => {
this.resubscribeAll();
});
await this.runReloadLoop();
}
destroy(): void {
this.destroyed = true;
if (this.flushTimer !== null) {
clearTimeout(this.flushTimer);
this.flushTimer = null;
}
this.offSubscriptionChange?.();
this.offSubscriptionChange = null;
// Flush any buffered events before tearing down.
if (this.buffer.length > 0) {
const toFlush = this.buffer.splice(0);
this.sendBatch(toFlush, "flush on destroy failed");
}
for (const [, unsub] of this.active) {
void unsub();
}
this.active.clear();
}
/**
* Request a reload of all save subscriptions.
*
* If no reload is currently running, starts one immediately via
* `runReloadLoop`. If one is already running, sets `reloadPending` so
* the loop runs exactly one additional full pass after the current pass
* completes — no matter how many changes arrive mid-run, they coalesce
* into a single follow-up pass.
*/
private resubscribeAll(): void {
if (this.reloading) {
this.reloadPending = true;
return;
}
void this.runReloadLoop();
}
/**
* Runs `doResubscribe` in a loop, draining any reload requests that
* arrive while a pass is in flight. The loop terminates once no reload
* is pending and the manager has not been destroyed.
*
* Awaited directly by `start()` for the initial load.
*/
private async runReloadLoop(): Promise<void> {
this.reloading = true;
try {
do {
this.reloadPending = false;
await this.doResubscribe();
} while (this.reloadPending && !this.destroyed);
} finally {
this.reloading = false;
}
}
/**
* One full reload pass: fetch the current subscription list, tear down
* removed or kind-changed subscriptions, and open new ones.
*
* Only one instance of this method ever runs at a time (enforced by
* `runReloadLoop`). That single-flight guarantee makes concurrent
* list/subscribe awaits structurally impossible.
*/
private async doResubscribe(): Promise<void> {
if (this.destroyed) return;
let subs: SaveSubscription[];
try {
subs = await this.deps.listSaveSubscriptions();
} catch (err) {
console.warn("[archiveSyncManager] list_save_subscriptions failed:", err);
return;
}
if (this.destroyed) return;
// Full keys (scope+kinds) for the current subscription list.
const wanted = new Set(
subs.map((s) => subKey(s.scopeType, s.scopeValue, s.kinds)),
);
// Tear down subscriptions that are no longer needed or whose kinds changed.
// A stale entry whose scope is still present but with different kinds will
// have a different full key and be absent from `wanted`, so it gets torn
// down here and recreated below with the new filter.
for (const [key, unsub] of this.active) {
if (!wanted.has(key)) {
void unsub();
this.active.delete(key);
}
}
// Open new subscriptions for any full key not already active.
// No concurrency guards are needed here: single-flight serialization
// ensures this loop body is the only async path running, so nothing
// can add a duplicate entry to `active` between iterations.
for (const sub of subs) {
if (this.destroyed) return;
const key = subKey(sub.scopeType, sub.scopeValue, sub.kinds);
if (this.active.has(key)) continue;
const scopeType = sub.scopeType;
const scopeValue = sub.scopeValue;
const filter = buildFilter(sub);
let dispose: (() => Promise<void>) | undefined;
try {
dispose = await this.deps.relayClient.subscribeLive(
filter,
(event: RelayEvent) => {
this.enqueue(event, scopeType, scopeValue);
},
);
} catch (err) {
console.warn(
`[archiveSyncManager] subscribeLive failed for ${scopeKey(scopeType, scopeValue)}:`,
err,
);
// Do NOT add key to active — next resubscribeAll will retry.
continue;
}
// A destroy() call may have arrived while subscribeLive was pending.
// Tear down the just-opened subscription and bail out.
if (this.destroyed) {
void dispose();
return;
}
this.active.set(key, dispose);
}
}
private enqueue(
event: RelayEvent,
scopeType: ScopeType,
scopeValue: string,
): void {
if (this.destroyed) return;
this.buffer.push({
rawEventJson: JSON.stringify(event),
matchedScope: { scopeType, scopeValue },
});
if (this.buffer.length >= this.flushBatchSize) {
this.flush();
} else {
this.scheduleFlush();
}
}
private scheduleFlush(): void {
if (this.flushTimer !== null) return;
this.flushTimer = setTimeout(() => {
this.flushTimer = null;
this.flush();
}, this.flushIdleMs);
}
private flush(): void {
if (this.flushTimer !== null) {
clearTimeout(this.flushTimer);
this.flushTimer = null;
}
if (this.buffer.length === 0) return;
const batch = this.buffer.splice(0);
this.sendBatch(batch, "archive_events failed");
}
/**
* Fire-and-forget `archiveEvents(batch)`, shared by the idle/size-triggered
* flush and the destroy-time flush. Notifies `onAgentMetricsChanged`
* subscribers only when the backend confirms `persistedAgentMetrics > 0` —
* the backend is authoritative, so a rejected call, a duplicate-only batch,
* or a batch with no kind-44200 events never notifies.
*/
private sendBatch(
batch: Array<{
rawEventJson: string;
matchedScope: { scopeType: ScopeType; scopeValue: string };
}>,
errLabel: string,
): void {
void this.deps
.archiveEvents(batch)
.then((result) => {
if (result.persistedAgentMetrics > 0) {
notifyAgentMetricsChanged();
}
})
.catch((err: unknown) => {
console.warn(`[archiveSyncManager] ${errLabel}:`, err);
});
}
}
// ── React hook ────────────────────────────────────────────────────────────────
import * as React from "react";
/**
* Starts the ArchiveSyncManager once `ready` is true and tears it down on
* unmount. The `ready` gate ensures observer reconciliation completes before
* any relay listeners open — kind 24200 is relay-ephemeral, so frames emitted
* before the listener opens are permanently lost.
*/
export function useArchiveSync(ready: boolean): void {
React.useEffect(() => {
if (!ready) return;
const manager = new ArchiveSyncManager();
void manager.start();
return () => {
manager.destroy();
};
}, [ready]);
}

View File

@@ -0,0 +1,24 @@
import * as React from "react";
import { listen } from "@tauri-apps/api/event";
import { notifyAgentMetricsChanged } from "@/shared/api/tauriArchive";
/**
* Bridges the backend's `archive-agent-metrics-changed` event to the existing
* in-process notifier.
*
* The archive batch that persists agent-metric rows now runs entirely in Rust,
* so the notifier's archive-side producer became a Tauri event. Consumers
* (`useAgentUsageSeries`) keep subscribing to `onAgentMetricsChanged` exactly
* as before — the source moved, the contract did not.
*/
export function useArchiveAgentMetricsBridge(): void {
React.useEffect(() => {
const unlisten = listen("archive-agent-metrics-changed", () => {
notifyAgentMetricsChanged();
});
return () => {
void unlisten.then((fn) => fn());
};
}, []);
}

View File

@@ -0,0 +1,334 @@
/**
* Mounted-hook tests for the archive sync start gate.
*
* The gate is the only part of archive sync still living in the renderer, and
* it is load-bearing for a reason no unit test of either hook alone can show:
* kind 24200 is relay-*ephemeral*. Frames that arrive before the listener
* opens are gone permanently, so `start_archive_sync` must not be invoked
* until observer reconciliation has seeded kind 24200 into the saved
* subscription. Everything below the gate now runs in Rust and is covered by
* `archive/sync_tests.rs`.
*
* These mount the two real hooks composed exactly as AppShell composes them
* (`useArchiveSync(useObserverArchiveReconciliation(pubkey))`) and assert on
* the Tauri commands that actually cross the boundary. They replace the
* ordering tests that drove the deleted `ArchiveSyncManager`: the invariant
* survived the move to Rust, only its observable did.
*/
import assert from "node:assert/strict";
import { afterEach, describe, it } from "node:test";
import { JSDOM } from "jsdom";
import React from "react";
import { act } from "react";
import { createRoot } from "react-dom/client";
import { useObserverArchiveReconciliation } from "./useObserverArchiveSeed.ts";
import { useArchiveSync } from "./useArchiveSync.ts";
const PUBKEY = "pk1";
/**
* Loads a fresh instance of the hook module — a new JS realm.
*
* The realm's epoch is memoized in module scope, which is exactly right in
* production (one realm announces once) and exactly what a reload destroys.
* Re-importing under a unique specifier reproduces that destruction honestly,
* rather than exporting a reset hook that exists only for tests.
*/
async function freshRealm() {
const mod = await import(`./useArchiveSync.ts?realm=${realmCounter++}`);
return mod.useArchiveSync;
}
let realmCounter = 0;
// ── Harness ──────────────────────────────────────────────────────────────────
/**
* Mounts the AppShell composition and returns the invoked Tauri command names
* in call order, plus handles to drive the gate and unmount.
*
* `@tauri-apps/api/core` reads `window.__TAURI_INTERNALS__.invoke` at call
* time, so intercepting it here captures every command the hooks issue.
*/
function mountGate({
mergeShouldFail = false,
windowLabel = "main",
useArchiveSync: useArchiveSyncImpl = useArchiveSync,
} = {}) {
const dom = new JSDOM(
"<!doctype html><html><body><div id='root'></div></body></html>",
);
const invoked = [];
const calls = [];
let epochs = 0;
dom.window.__TAURI_INTERNALS__ = {
invoke: (cmd, args) => {
invoked.push(cmd);
calls.push({ cmd, epoch: args?.epoch, lease: args?.lease });
if (cmd === "announce_archive_sync_epoch") {
epochs += 1;
return Promise.resolve(epochs);
}
return Promise.resolve(null);
},
// `getCurrentWindow()` reads exactly this path (api/window.js:85), and
// `isTauri()` reads `globalThis.isTauri` (api/core.js:280). Both are what
// `huddleWindowChannelId` uses to tell a companion realm from the main one.
metadata: { currentWindow: { label: windowLabel } },
transformCallback: () => Math.random(),
};
Object.assign(globalThis, {
isTauri: true,
document: dom.window.document,
HTMLElement: dom.window.HTMLElement,
IS_REACT_ACT_ENVIRONMENT: true,
window: dom.window,
});
let resolveMerge;
let rejectMerge;
const mergePromise = new Promise((resolve, reject) => {
resolveMerge = resolve;
rejectMerge = () => reject(new Error("merge failed"));
});
// Real reconciler, controllable merge: this is the production seam the
// reconciliation hook already exposes for tests. Frozen so the hook's
// effect does not re-run on every render.
const deps = Object.freeze({
mergeSaveSubscriptionKinds: () => mergePromise,
readExplicitChoice: () => "unset",
setExplicitChoice: () => {},
});
function Harness() {
const reconciled = useObserverArchiveReconciliation(PUBKEY, deps);
useArchiveSyncImpl(reconciled);
return null;
}
const root = createRoot(dom.window.document.getElementById("root"));
return {
invoked,
calls,
async mount() {
await act(async () => {
root.render(React.createElement(Harness));
});
},
async settleGate() {
await act(async () => {
if (mergeShouldFail) rejectMerge();
else resolveMerge();
await mergePromise.catch(() => {});
});
},
async unmount() {
await act(async () => {
root.unmount();
});
},
};
}
afterEach(() => {
delete globalThis.isTauri;
delete globalThis.document;
delete globalThis.window;
delete globalThis.HTMLElement;
delete globalThis.IS_REACT_ACT_ENVIRONMENT;
});
// ── The gate ─────────────────────────────────────────────────────────────────
describe("archive sync start gate", () => {
it("does not start the backend task before reconciliation resolves", async () => {
const gate = mountGate();
await gate.mount();
assert.deepEqual(
gate.invoked.filter((cmd) => cmd === "start_archive_sync"),
[],
"start_archive_sync must not run before reconciliation resolves — " +
"kind 24200 frames arriving before the listener opens are lost",
);
await gate.settleGate();
assert.deepEqual(
gate.invoked.filter((cmd) => cmd === "start_archive_sync"),
["start_archive_sync"],
"start_archive_sync must run exactly once after the gate opens",
);
});
it("never starts the backend task when reconciliation fails", async () => {
const gate = mountGate({ mergeShouldFail: true });
await gate.mount();
await gate.settleGate();
assert.deepEqual(
gate.invoked.filter((cmd) => cmd === "start_archive_sync"),
[],
"a failed reconciliation leaves the gate closed",
);
});
it("stops the backend task on unmount", async () => {
const gate = mountGate();
await gate.mount();
await gate.settleGate();
await gate.unmount();
assert.deepEqual(
gate.invoked.filter((cmd) => cmd.endsWith("_archive_sync")),
["start_archive_sync", "stop_archive_sync"],
"unmount must stop the task it started, in that order",
);
});
});
// ── Lifecycle ownership ──────────────────────────────────────────────────────
//
// Both commands are fired without awaiting, and Tauri completes them in an
// unconstrained order. The lease is what lets the backend recover the
// renderer's intent order from calls that arrive in any order, so what these
// assert is that the renderer allocates it correctly — synchronously, in effect
// order, with each cleanup carrying its own start's lease.
//
// The backend half of the invariant (a stale lease changes nothing) is proven
// in `archive/sync_tests.rs`, where the completion order can be chosen
// directly. Neither test can stand alone: this one cannot control backend
// ordering, and that one cannot show which lease the renderer sent.
describe("archive sync lifecycle leases", () => {
it("pairs each stop with the lease of the start it is cleaning up", async () => {
const gate = mountGate();
await gate.mount();
await gate.settleGate();
await gate.unmount();
const lifecycle = gate.calls.filter(({ cmd }) =>
cmd.endsWith("_archive_sync"),
);
assert.equal(lifecycle.length, 2, "expected one start and one stop");
const [start, stop] = lifecycle;
assert.equal(typeof start.lease, "number", "start must carry a lease");
assert.equal(
stop.lease,
start.lease,
"cleanup must carry its own start's lease — a stop that sends anything " +
"else either cancels a newer owner's task or no-ops its own",
);
});
it("allocates a strictly newer lease for each remount", async () => {
const first = mountGate();
await first.mount();
await first.settleGate();
await first.unmount();
const second = mountGate();
await second.mount();
await second.settleGate();
const leaseOf = (gate, cmd) =>
gate.calls.find((call) => call.cmd === cmd)?.lease;
assert.ok(
leaseOf(second, "start_archive_sync") >
leaseOf(first, "start_archive_sync"),
"a remount's start must outrank the previous mount's, or the backend " +
"cannot tell which start the app actually wants",
);
assert.ok(
leaseOf(second, "start_archive_sync") >
leaseOf(first, "stop_archive_sync"),
"a remount's start must also outrank the previous cleanup, so a stop " +
"that lands late cannot strand the new task",
);
});
});
// ── Realm ownership ──────────────────────────────────────────────────────────
//
// `AppShell` is the root route, so a companion huddle window mounts this same
// tree in a second JS realm. Archive sync is app-global work and the main
// window owns it, exactly as it owns microphone capture. The epoch cannot fix
// this: epochs order realms in TIME, and a companion is a second realm in
// SPACE — its unmount cleanup would carry the newest epoch and cancel the live
// main-window task. So the companion must never announce or issue lifecycle
// commands at all.
describe("archive sync realm ownership", () => {
it("issues no archive lifecycle commands from a companion window", async () => {
const gate = mountGate({
windowLabel: "huddle-11111111-2222-3333-4444-555555555555",
});
await gate.mount();
await gate.settleGate();
await gate.unmount();
assert.deepEqual(
gate.invoked.filter(
(cmd) =>
cmd.endsWith("_archive_sync") ||
cmd === "announce_archive_sync_epoch",
),
[],
"a companion realm must not announce or run lifecycle commands — its " +
"cleanup would cancel the main window's live sync task",
);
});
it("still runs the lifecycle in the main window", async () => {
const gate = mountGate({ windowLabel: "main" });
await gate.mount();
await gate.settleGate();
await gate.unmount();
assert.deepEqual(
gate.invoked.filter((cmd) => cmd.endsWith("_archive_sync")),
["start_archive_sync", "stop_archive_sync"],
"the owning window must be unaffected by the companion exclusion",
);
});
it("resolves the epoch before issuing any lifecycle command", async () => {
const gate = mountGate({ useArchiveSync: await freshRealm() });
await gate.mount();
await gate.settleGate();
await gate.unmount();
const relevant = gate.invoked.filter(
(cmd) =>
cmd.endsWith("_archive_sync") || cmd === "announce_archive_sync_epoch",
);
assert.equal(
relevant[0],
"announce_archive_sync_epoch",
"the epoch must be announced before any lifecycle command — an " +
"unawaited announcement is just another racing invoke",
);
const lifecycle = gate.calls.filter(({ cmd }) =>
cmd.endsWith("_archive_sync"),
);
for (const call of lifecycle) {
assert.equal(
typeof call.epoch,
"number",
`${call.cmd} must carry the resolved epoch`,
);
}
assert.equal(
lifecycle[0].epoch,
lifecycle[1].epoch,
"start and its cleanup must share one realm epoch",
);
});
});

View File

@@ -0,0 +1,96 @@
import * as React from "react";
import { huddleWindowChannelId } from "@/features/huddle/lib/huddleWindow";
import {
announceArchiveSyncEpoch,
nextArchiveSyncLease,
startArchiveSync,
stopArchiveSync,
} from "@/shared/api/tauriArchive";
/**
* This realm's epoch, announced once at first use.
*
* Per-realm, NOT per-effect: the epoch's whole job is to order successive
* realms, so minting one per effect would make it order announcements instead
* and leave remounts inside a realm racing exactly as before. Memoizing the
* promise also means a remount cannot overtake its own predecessor's
* announcement.
*/
let realmEpoch: Promise<number> | null = null;
function archiveSyncEpoch(): Promise<number> {
// Clear on failure so a remount can retry; a cached rejection would make one
// transient IPC error permanent for the life of the realm.
realmEpoch ??= announceArchiveSyncEpoch().catch((err: unknown) => {
realmEpoch = null;
throw err;
});
return realmEpoch;
}
/**
* Starts the native archive sync task once `ready` is true and stops it on
* unmount.
*
* The `ready` gate is load-bearing and cannot move into the backend: kind
* 24200 is relay-ephemeral, so frames emitted before the listener opens are
* permanently lost. Observer reconciliation must have seeded kind 24200 into
* the saved subscription before any listener opens, and only the renderer
* knows when that finished. The backend task is therefore not self-starting.
*
* Everything below this call — subscribing, buffering, batching, archiving —
* now runs in Rust; the renderer sees no archive traffic at all.
*
* Ownership is main-window-only. `AppShell` is the root route, so a companion
* window (`huddle::window`) mounts this same tree in a second JS realm — and
* archive sync is app-global work, exactly like the microphone capture the
* main window already owns. Two concurrent realms cannot be ordered by a
* newest-wins clock: the companion's unmount cleanup would cancel the live
* main-window task while the main realm sits there with nothing to re-issue.
* So secondary realms never announce and never issue lifecycle commands.
*
* Within the owning window, `(epoch, lease)` orders everything else: the epoch
* is minted by Rust and survives renderer reload (a reload resets the JS lease
* counter but not the backend's mark), and the lease orders effects inside one
* realm. The epoch must be resolved BEFORE any lifecycle command is sent — an
* unawaited announcement is just another racing invoke.
*/
export function useArchiveSync(ready: boolean): void {
React.useEffect(() => {
if (!ready) return;
// Companion realms do not participate; see the ownership rule above.
if (huddleWindowChannelId() !== null) return;
const lease = nextArchiveSyncLease();
let stopped = false;
const started = archiveSyncEpoch()
.then(async (epoch) => {
// The cleanup may have run while the epoch was in flight. Issuing the
// start now would install a task nobody owns, so record the epoch and
// let the cleanup below stop it under the same mark.
if (stopped) return epoch;
await startArchiveSync(epoch, lease);
return epoch;
})
.catch((err: unknown) => {
console.warn("[useArchiveSync] start_archive_sync failed:", err);
return null;
});
return () => {
stopped = true;
// Chained off the same promise so the stop cannot overtake its start:
// the epoch is not knowable until the announcement resolves.
void started
.then(async (epoch) => {
if (epoch === null) return;
await stopArchiveSync(epoch, lease);
})
.catch((err: unknown) => {
console.warn("[useArchiveSync] stop_archive_sync failed:", err);
});
};
}, [ready]);
}

View File

@@ -6,7 +6,6 @@ import {
reconcileObserverArchive,
startReconciliation,
} from "./useObserverArchiveSeed.ts";
import { ArchiveSyncManager } from "./archiveSyncManager.ts";
// ── Fake deps factory ────────────────────────────────────────────────────────
@@ -140,117 +139,12 @@ test("test_reconcile_toggle_off_then_restart_does_not_remerge", async () => {
);
});
// ── Startup ordering (real ArchiveSyncManager + real reconciler) ─────────────
test("test_archive_sync_blocked_until_reconciliation", async () => {
let resolveMerge;
const mergePromise = new Promise((resolve) => {
resolveMerge = resolve;
});
const subscribeCalls = [];
const fakeRelay = {
subscribeLive(filter, _callback) {
subscribeCalls.push(filter);
return Promise.resolve(async () => {});
},
};
const reconcilerDeps = {
mergeSaveSubscriptionKinds: () => mergePromise,
readExplicitChoice: () => "unset",
setExplicitChoice: () => {},
};
const manager = new ArchiveSyncManager({
relayClient: fakeRelay,
listSaveSubscriptions: async () => [
{
scopeType: "owner_p",
scopeValue: "pk1",
kinds: [24200],
identityPubkey: "pk1",
relayUrl: "wss://r",
createdAt: 0,
},
],
archiveEvents: async () => ({ persisted: 0, dropped: 0 }),
onSubscriptionChange: () => () => {},
});
// Start reconciliation (pending — merge not yet resolved).
const reconciling = reconcileObserverArchive("pk1", reconcilerDeps);
// Before reconciliation resolves, manager must not have been started.
await tick();
assert.equal(
subscribeCalls.length,
0,
"subscribeLive must not run before reconciliation",
);
// Resolve reconciliation — now start the manager (simulating the gate).
resolveMerge();
await reconciling;
await manager.start();
assert.ok(
subscribeCalls.length > 0,
"subscribeLive must run after gate opens",
);
const hasOwnerP = subscribeCalls.some((f) => f["#p"]?.length > 0);
assert.ok(hasOwnerP, "subscription must use owner_p (#p) filter");
const hasKind24200 = subscribeCalls.some((f) => f.kinds?.includes(24200));
assert.ok(hasKind24200, "subscription filter must include kind 24200");
manager.destroy();
});
test("test_archive_sync_blocked_on_reconciliation_rejection", async () => {
const reconcilerDeps = makeDeps({ mergeShouldFail: true });
const subscribeCalls = [];
const fakeRelay = {
subscribeLive(filter) {
subscribeCalls.push(filter);
return Promise.resolve(async () => {});
},
};
const manager = new ArchiveSyncManager({
relayClient: fakeRelay,
listSaveSubscriptions: async () => [
{
scopeType: "owner_p",
scopeValue: "pk1",
kinds: [24200],
identityPubkey: "pk1",
relayUrl: "wss://r",
createdAt: 0,
},
],
archiveEvents: async () => ({ persisted: 0, dropped: 0 }),
onSubscriptionChange: () => () => {},
});
// Reconciliation rejects — gate must remain closed.
let rejected = false;
try {
await reconcileObserverArchive("pk1", reconcilerDeps);
} catch {
rejected = true;
}
assert.ok(rejected, "reconciliation must reject on merge failure");
// Manager must NOT start after failed reconciliation.
assert.equal(
subscribeCalls.length,
0,
"subscribeLive must not run after failed reconciliation",
);
manager.destroy();
});
// ── Startup ordering ─────────────────────────────────────────────────────────
//
// The two ordering tests that lived here drove `ArchiveSyncManager` directly.
// That manager is gone: archive sync runs in Rust and the renderer keeps only
// the start gate. The same invariant — no listener opens before kind 24200 is
// seeded — is now asserted against the real gate in useArchiveSync.test.mjs.
// ── Identity-scoped readiness (exercises exported isReconciledFor) ──────────

View File

@@ -158,34 +158,17 @@ function decodeArchiveBatchResult(
};
}
// ── Subscription-change notifier ─────────────────────────────────────────────
/**
* Module-level notifier for subscription mutations (create/delete).
* The archive sync manager subscribes to this to reload live subscriptions
* without needing manager instances threaded through UI props.
*/
const subscriptionChangeListeners = new Set<() => void>();
export function onSubscriptionChange(listener: () => void): () => void {
subscriptionChangeListeners.add(listener);
return () => subscriptionChangeListeners.delete(listener);
}
function notifySubscriptionChange(): void {
for (const listener of subscriptionChangeListeners) {
listener();
}
}
// ── Agent-metrics-change notifier ────────────────────────────────────────────
/**
* Module-level notifier for newly persisted agent turn metrics (kind 44200).
* `useAgentUsageSeries` subscribes to this to invalidate its query without
* polling. Fired only when the backend confirms `persistedAgentMetrics > 0`
* for a successful `archiveEvents` call, or when a kind-44200 subscription
* mutation succeeds (`collectionEnabled` is part of the usage query result).
* polling. Two producers: a kind-44200 subscription mutation succeeding here
* (`collectionEnabled` is part of the usage query result), and the native
* archive sync task persisting new metric rows — that one arrives as the
* `archive-agent-metrics-changed` Tauri event, bridged by
* `useArchiveAgentMetricsBridge`, since the batch it belongs to no longer
* passes through JS.
*/
const agentMetricsChangeListeners = new Set<() => void>();
@@ -267,7 +250,6 @@ export async function agentMetricArchiveDefaultEnabled(): Promise<boolean> {
*/
export async function mergeSaveSubscriptionKinds(kind: number): Promise<void> {
await invokeTauri("merge_save_subscription_kinds", { kind });
notifySubscriptionChange();
// `collectionEnabled` is part of the usage query result — toggling kind
// 44200 on must invalidate mounted usage queries. Other kinds don't affect
// usage state.
@@ -291,7 +273,6 @@ export async function mergeSaveSubscriptionKinds(kind: number): Promise<void> {
*/
export async function removeSaveSubscriptionKind(kind: number): Promise<void> {
await invokeTauri("remove_save_subscription_kind", { kind });
notifySubscriptionChange();
if (kind === KIND_AGENT_TURN_METRIC) {
notifyAgentMetricsChanged();
}
@@ -312,7 +293,71 @@ export async function createSaveSubscription(
scopeValue,
kinds,
});
notifySubscriptionChange();
}
// ── Native archive sync lifecycle ────────────────────────────────────────────
/**
* Monotonic lease counter for archive-sync lifecycle commands.
*
* Allocated synchronously in the renderer, before `invoke`, because effect
* execution order IS intent order and it is the only place that ordering is
* free. Tauri commands complete in an unconstrained order, so a token minted by
* the backend records which call reached the mutex first — not which the app
* actually wants. A remount whose `start` is delayed past the newer effect's
* would otherwise mint the newest token for the stalest caller, whose cleanup
* then holds a valid warrant to cancel the live task.
*
* This counter is realm-scoped: it resets to zero whenever the renderer
* reloads, while the backend's mark persists for the life of the Tauri
* process. The epoch below is what orders those successive realms.
*/
let archiveSyncLease = 0;
/** Allocates the next lease. Call synchronously in effect order. */
export function nextArchiveSyncLease(): number {
archiveSyncLease += 1;
return archiveSyncLease;
}
/**
* Announces this renderer realm and resolves its epoch.
*
* Must be awaited before any lifecycle command is issued: an unawaited
* announcement is just another racing `invoke`, which would order
* announcements rather than realms and reintroduce the arrival-order bug one
* level up.
*
* Only the main window announces. Archive sync is app-global and
* main-window-owned — the same rule that keeps the main window the owner of
* microphone capture. Epochs order realms in time; a companion window is a
* second realm in space, and a newest-wins clock cannot model two concurrent
* owners (the companion's cleanup would cancel the live main-window task).
*/
export async function announceArchiveSyncEpoch(): Promise<number> {
return await invokeTauri<number>("announce_archive_sync_epoch");
}
/**
* Start the backend archive sync task for the current identity.
*
* Idempotent per identity + relay. Must only be called after observer
* reconciliation resolves — see `useArchiveSync` for why the backend cannot
* gate itself. The backend ignores a mark older than the newest it has seen.
*/
export async function startArchiveSync(
epoch: number,
lease: number,
): Promise<void> {
await invokeTauri("start_archive_sync", { epoch, lease });
}
/** Stop the backend archive sync task. Ignored if `(epoch, lease)` is stale. */
export async function stopArchiveSync(
epoch: number,
lease: number,
): Promise<void> {
await invokeTauri("stop_archive_sync", { epoch, lease });
}
/**
@@ -338,9 +383,6 @@ export async function deleteSaveSubscription(
scopeType,
scopeValue,
});
if (removed) {
notifySubscriptionChange();
}
return removed;
}

View File

@@ -13358,6 +13358,24 @@ export function maybeInstallE2eTauriMocks() {
case "archive_events":
// Returns the ArchiveBatchResult shape the UI expects.
return { persisted: 0, dropped: 0 };
// Archive sync runs natively; the bridge has no relay-backed backend to
// drive, so these are accepted no-ops. Without them every AppShell mount
// logs an unknown-command warning once the gate opens.
//
// The epoch must return a number rather than null, because it stands in
// for a value production guarantees is numeric. Nothing here validates
// it at runtime — the typed wrapper is erased — so a null would not
// fail loudly; it would flow into `start_archive_sync` as
// `{ epoch: null }` and this no-op would accept it, leaving the suite
// exercising a wire shape the real backend never sends. Returning a
// number keeps the production contract, and the reconciliation-gate
// spec asserts it: announce precedes start, and start carries a numeric
// epoch. Mutating this to null fails that spec.
case "announce_archive_sync_epoch":
return 1;
case "start_archive_sync":
case "stop_archive_sync":
return null;
case "agent_metric_archive_default_enabled":
return activeConfig?.mock?.agentMetricArchiveDefaultEnabled ?? true;
case "set_prevent_sleep_active":

View File

@@ -118,12 +118,31 @@ test.describe("observer archive policy — Settings toggle", () => {
});
test.describe("observer archive policy — reconciliation gate", () => {
test("archive sync reaches subscription path after reconciliation", async ({
test("archive sync starts only after reconciliation seeds the subscription", async ({
page,
}) => {
// The reconciliation gate (useObserverArchiveReconciliation) must resolve
// successfully for a fresh identity, allowing useArchiveSync to start the
// ArchiveSyncManager, which calls list_save_subscriptions.
// The gate invariant: no archive listener may open before observer
// reconciliation has seeded kind 24200. Kind 24200 is relay-ephemeral, so
// a listener opened early misses frames permanently.
//
// The archive subscription and its `#p` + kind-24200 REQ now live entirely
// in Rust, so this spec asserts the gate at the seam that is still in JS:
// the `start_archive_sync` start signal, which the renderer emits only
// after reconciliation resolves. Ownership of the invariants this test
// used to cover:
//
// REQ shape (`#p` tag key, live tail) →
// native_relay_client::relay_backed_tests::
// archive_sync_session_receives_live_events_from_a_real_relay
// (relay-backed; `#p`→`#e` negative control confirms it is bound to
// the tag key)
// gate closed on pending/failed reconciliation →
// src/features/local-archive/useArchiveSync.test.mjs
// (mutation-verified: deleting the gate fails 3/3 there)
//
// Deliberately NOT re-asserted through the mock bridge: the bridge has no
// Rust task to drive, so making it synthesize the REQ would assert the
// mock's own re-implementation rather than the shipped behavior.
await installMockBridge(page, {
saveSubscriptions: [
{
@@ -141,42 +160,67 @@ test.describe("observer archive policy — reconciliation gate", () => {
timeout: 10_000,
});
// The IPC counter proves the subscription path was reached.
await page.waitForFunction(
() => {
const counters = (window as Record<string, unknown>)
.__BUZZ_E2E_IPC_COUNTERS__ as Record<string, number> | undefined;
return (counters?.list_save_subscriptions ?? 0) > 0;
},
null,
{ timeout: 10_000 },
);
const readCommands = () =>
page.evaluate(
() =>
(window as Window & { __BUZZ_E2E_COMMANDS__?: string[] })
.__BUZZ_E2E_COMMANDS__ ?? [],
);
const count = await page.evaluate(() => {
const counters = (window as Record<string, unknown>)
.__BUZZ_E2E_IPC_COUNTERS__ as Record<string, number> | undefined;
return counters?.list_save_subscriptions ?? 0;
});
expect(count).toBeGreaterThan(0);
await expect
.poll(async () => (await readCommands()).includes("start_archive_sync"), {
timeout: 10_000,
})
.toBe(true);
// The reconciliation gate must also result in a real `#p` + kind-24200
// live REQ filter.
const hasOwnerKindSubscription = await page.evaluate(
(ownerPubkey) =>
const commands = await readCommands();
// What this e2e uniquely owns is WIRING: that AppShell actually mounts
// `useArchiveSync` and its start signal reaches the IPC boundary. Drop
// the hook from AppShell and archive never starts in the shipped app,
// and no unit test notices — they mount the hook directly.
//
// It deliberately does NOT assert gate ORDERING. The bridge records a
// command when it is invoked, not when it resolves, and reconciliation's
// `merge_save_subscription_kinds` is invoked synchronously on mount — so
// `merge` precedes `start` in this log even with the `ready` gate deleted.
// Verified by mutation: removing the gate leaves this spec 6/6 green while
// `useArchiveSync.test.mjs` fails 3/3. The gate is asserted there, where
// the assertion has teeth; asserting it here would only look like coverage.
expect(commands).toContain("merge_save_subscription_kinds");
expect(commands).toContain("start_archive_sync");
// The epoch handshake, asserted on the wire rather than by construction.
// The renderer must announce and AWAIT its realm epoch before it may issue
// any lifecycle command, and must pass that epoch through. Nothing here
// validates types at runtime — the typed wrapper is erased — so a bridge
// returning null would flow null into `startArchiveSync(null, lease)` and
// this no-op bridge would accept it. Asserting only that the command was
// named passes either way; asserting order and payload is what makes this
// a receipt.
const payloads = await page.evaluate(
() =>
(
window as Window & {
__BUZZ_E2E_HAS_MOCK_OWNER_KIND_SUBSCRIPTION__?: (input: {
ownerPubkey: string;
kind: number;
}) => boolean;
__BUZZ_E2E_COMMAND_PAYLOADS__?: Array<{
command: string;
payload: unknown;
}>;
}
).__BUZZ_E2E_HAS_MOCK_OWNER_KIND_SUBSCRIPTION__?.({
ownerPubkey,
kind: 24200,
}) ?? false,
"deadbeef".repeat(8),
).__BUZZ_E2E_COMMAND_PAYLOADS__ ?? [],
);
expect(hasOwnerKindSubscription).toBe(true);
const announcedAt = payloads.findIndex(
(entry) => entry.command === "announce_archive_sync_epoch",
);
const startedAt = payloads.findIndex(
(entry) => entry.command === "start_archive_sync",
);
expect(announcedAt).toBeGreaterThanOrEqual(0);
expect(startedAt).toBeGreaterThan(announcedAt);
expect(payloads[startedAt]?.payload).toMatchObject({
epoch: expect.any(Number),
lease: expect.any(Number),
});
});
test("fresh install with empty subscriptions: reconciliation seeds kind 24200", async ({
@@ -193,24 +237,36 @@ test.describe("observer archive policy — reconciliation gate", () => {
timeout: 10_000,
});
// The seeded row itself, read back through `list_save_subscriptions` —
// the state the repair path is supposed to produce. This replaces a poll
// on the live `#p`+24200 REQ: that REQ is now opened by the Rust archive
// task, which the mock bridge never runs. Seeding is unchanged by that
// move and stays asserted here.
await expect
.poll(
() =>
page.evaluate(
(ownerPubkey) =>
(
window as Window & {
__BUZZ_E2E_HAS_MOCK_OWNER_KIND_SUBSCRIPTION__?: (input: {
ownerPubkey: string;
kind: number;
}) => boolean;
}
).__BUZZ_E2E_HAS_MOCK_OWNER_KIND_SUBSCRIPTION__?.({
ownerPubkey,
kind: 24200,
}) ?? false,
"deadbeef".repeat(8),
),
page.evaluate(async (ownerPubkey) => {
const invoke = (
window as Window & {
__BUZZ_E2E_INVOKE_MOCK_COMMAND__?: (
command: string,
payload?: unknown,
) => Promise<unknown>;
}
).__BUZZ_E2E_INVOKE_MOCK_COMMAND__;
if (!invoke) return false;
const rows = (await invoke("list_save_subscriptions")) as Array<{
scope_type: string;
scope_value: string;
kinds: string;
}>;
return rows.some(
(row) =>
row.scope_type === "owner_p" &&
row.scope_value === ownerPubkey &&
(JSON.parse(row.kinds) as number[]).includes(24200),
);
}, "deadbeef".repeat(8)),
{ timeout: 10_000 },
)
.toBe(true);