mirror of
https://github.com/block/buzz.git
synced 2026-10-27 12:40:30 +00:00
fix(desktop): bound startup request bursts and recover quota refusals (#7790)
Prepared by Brain on Wes’s behalf. ## Summary Reduce startup request amplification and let quota-refused background work recover without changing relay quotas or bypassing the existing Send gate. - Coalesce nest regeneration into one worker plus the latest pending generation. Keep the stale-write fence; bound the archive read so an old workspace cannot strand newer work. - Claim pending channel subscriptions before awaiting setup; cancel removed/unmounted entries and preserve still-owned subscriptions across reconnect handoff. - Recheck extended cooldown deadlines before retry dispatch; distinguish an explicit zero-second hint from a missing hint. - Recover native finite/unread queries within their original caller deadline and a three-retry budget, preserving verified unique events, complete filters and terminal errors. - Pace renderer initial/live-CLOSED REQs at one per 250ms: visible channel and explicitly interactive TTS/read-state/observer owners first, then live-only auxiliaries, then background history. Publication and finite queries bypass this queue. Reconnect replay now promotes explicit interactive owners alongside the visible channel with stable registration-order ties; its existing capped batches and cooldown remain intact. **Scope at `867495426`:** 39 source/test files, including 15 production files (+665/-235) and 24 test/fixture files. The diagnostic recorder, HTTP instrumentation, trace launcher and raw captures are deliberately excluded. No profile reset, grouping, quota increase or new global scheduler. ## Before / after evidence These are observations from instrumented, human-operated development Tauri/Vite runs against a production relay, plus explicitly identified controlled regressions—not a packaged-release startup benchmark. | Measurement | Before | After | Evidence boundary | | --- | --- | --- | --- | | Archive snapshot requests (kind 13535) | 290 | 3 | Recorder-only baseline → nest-only intermediate candidate; unequal 67.436s / 39.273s captures and hydration differences | | Archive responses: HTTP 200 / 429 | 277 / 13 | 3 / 0 | Same baseline/nest-only captures; not a claim that all quota pressure disappeared | | Successful archive responses’ advertised content-length sum | 21,208,505 bytes | 229,695 bytes | Header values, not actual transferred bytes | | 154-channel StrictMode replay | 308 REQs | 154 REQs | Controlled real React hook → real session → mocked Tauri transport regression | | Add one channel while 154 setups are pending | 309 REQs | 155 REQs | Same production-boundary regression; ordinary mount remains 154 → 154 | | Broad renderer request burst / spacing | 110 broad REQs within 10.766ms in pre-drain capture | 250.061ms minimum / 252.131ms median between broad REQ writes in final longer capture | Different attended runs; pacing rather than a universal request-rate cap | | Native unread query outcomes | Baseline: 8 EOSE / 146 quota CLOSED out of 154 queries | Final longer run: all 156 observed queries reach EOSE; eight refused queries each recover after two refusals | Different runs/workloads; query completion, not badge rendering | ### Timing metrics | Observation | Measured value | | --- | ---: | | Baseline’s two accepted chat EVENT socket writes → positive OK | 173.340ms / 85.556ms | | Nest-only intermediate candidate’s two accepted writes → positive OK | 277.854ms / 78.775ms | | Final candidate, short run, accepted chat writes → positive OK | 80.636ms / 77.506ms | | Final candidate, longer run, accepted chat writes → positive OK | 120.712ms / 86.895ms | | Last initial EOSE among 163 observed broad channel registrations, longer run | 64.550s after trace start | | Last EOSE among 156 observed native unread queries, longer run | 73.972s after trace start | | Native retry waits after 3s then 0s hints, longer run | 3000.7–3002.8ms, then 2001.5–2002.6ms | | Renderer finite-query retry after a 2s hint, longer run | 2061.365ms | Both Sends in each final run were accepted, with no recorded negative OK. Every quota-refused query in the longer run subsequently completed the same query. Its two Sends happened early (17.878s and 21.655s), not near the end. These small samples do not support a percentage latency improvement. **Tradeoffs and limits** - 296 background subscriptions imply about 74s of paced registration before extra cooldowns/high-priority work. The longer capture observed 163 registrations; it did not independently enumerate the expected membership set or record a queue-empty marker. - This is not a global 4-REQ/s cap: finite queries, native requests and reconnect replay are outside the drain and still share relay quota. Quota refusals recover rather than disappear. - Socket-write → OK starts after composer submission and any publisher gate wait. No click-to-visible, launch-to-usable or badge-paint latency is claimed. - Native EOSE is not proof of command return, classification, React state application or visible unread badges. Existing per-channel kinds, since floors, limits, h-less routing, unread lease/scope guard and eight-query concurrency remain intact. - The attended measurements belong to the instrumented candidate, not a fresh release run of this fixes-only commit. At the initial packaging checkpoint (`1bf3d060`), eighteen of twenty files were byte-identical; the other two removed diagnostic gate calls and isolated the nest test’s temporary paths. The subsequent interactive priority/reconnect and readiness fixes have controlled hook/session/browser evidence, not another attended voice/performance run. ### Evidence provenance (captures retained locally, not uploaded) | Capture | SHA-256 | | --- | --- | | Baseline `STARTUP_20260921T194814Z_98811.jsonl` | `f4f0a89ad9aea91084bd6a7432510488166de4cbc906600d88c26d557f380994` | | Nest-only `STARTUP_20260921T202610Z_28642.jsonl` | `ce517b02a3f717c36d96b36bb8616e8fa80b03b12b356bacfd40475925ea3f3e` | | Final short `STARTUP_20260922T013628Z_13506.jsonl` | `bfea6f3cf80cd7b7e4e676632a3df2f16271fade5bfa0edec72b7bcbbe7839ff` | | Final longer `STARTUP_20260922T014637Z_15846.jsonl` | `f4abece613d108c4c2fe3cb808e633a0486bdc9993ca92323822dc00c3fae6ec` | Final instrumented build fingerprint: `b0b4fc27d793c3e643f4e10289b5afdb4620ee2e13a8b33fa0595dd3417fd5ca`. All observed dropped-record counters in these final captures were zero. Source manifests and offline correlation analyses are retained locally; no raw captures, profile data or agent sessions are part of this PR. ## Validation Current head: `86749542657a60abdfe82eed5cc576d3e162fe5d`, based on `77729abfb692b25a0f4ec4a69add86af2e32c0dd`. Review follow-ups: - `8842dbd8`: explicit interactive TTS admission using the existing main-window owner, independent of companion/timeline selection. Real-hook/session/IPC coverage with 296 cold owners proves initial and CLOSED priority, unchanged filters/cooldown/pacing, background progress and deduplicated speech. - `8cb088106`: reconnect preserves interactive priority even with no visible channel. Stable ordering among foreground ties; real hook/session reconnect regressions cover cooldown, capped batching, all 297 owners, deduplication and unmount cancellation. Independent reviewer mutation restored the prior replay implementation, failed both reconnect rows, then passed the restored candidate. - `68e76d7c`: read-state and observer reuse interactive priority. Stop and ModelPicker wait on the existing observer owner inside send callbacks, after installing result listeners/deadlines. No new owner, timer or scheduler; inherited CLOSED/fallback/error semantics remain, rather than a new strict EOSE guarantee. The real Stop panel browser probe holds observer IPC across the click and checks observer-before-control order and correlated outcome. ModelPicker component ordering is source-reviewed only. - Mock live admission now replays matching stored history with per-filter time/limit constraints, deduplication, untagged auxiliary channel fallback and authoritative explicit h-tags. Kind-specific fixture admission waits and retry-resolving hover actions fix startup-dependent test setup without weakening event/badge assertions. - `48e9c0dad` and `867495426`: steady-state scroll-prepend setup observes live admission, subsequent head dispatch, successful/idle query completion and initial bottom geometry. Phase-2 genuine-growth and ≤32px anchor assertions are unchanged. This isolates steady-state prepending: it **does not fix or certify production early-scroll/head-reset behavior**. Current validation: - Normal push hooks at `867495426`: **6,582/6,582 renderer tests**, zero failed/skipped/cancelled; lint/static checks, typecheck and differential file-size gate passed. Native source is unchanged from the earlier checkpoint; no new native run is claimed for these renderer/test follow-ups. - E2E build passed at `48e9c0dad` plus the exact subsequently committed test-only delta (spec blob `469be2eed868c677ba80b859cc871208000fee60`). The completed affected-smoke run passed **153/153** across agent-control, badge, entity links, messaging, profile status and scroll history; the full scroll file additionally passed twice (**36/36**). No full CI-smoke pass is claimed from this six-file local run. - Pinky independently reviewed production and test follow-ups, including the final scroll spec blob `469be2eed868c677ba80b859cc871208000fee60`, with no blocking finding. Review is source evidence, not an independent test run. - **Not merge-ready:** new-head required CI must complete, exact-range security review requires authorization, and GitHub currently reports REVIEW_REQUIRED; fresh exact-head reviewer approval is still needed. No approval, dismissal or merge performed. No new packaged/native audio, launch-to-usable or badge-paint observation is claimed. Initial fixes-only checkpoint validation (`1bf3d06024d284a63326fdd353baced0842aeb81`): - Pre-commit formatting hooks passed; worktree clean. - Renderer package hooks: 6,570 tests passed, zero failed/skipped/cancelled; lint/static checks, TypeScript and differential file-size gate passed. - Native package hook passed: default and mesh-enabled all-target Clippy, then the complete native workspace test suite. App library: 3,222 passed / 0 failed / 19 ignored; terminal/integration suites passed. The real nest trigger/stalled-body test passed on macOS. - Independent whole-change source review: no blocking findings at this commit; all 20 file hashes matched the committed blobs and review/validation manifests. - Regression coverage includes real hook/session cancellation, StrictMode/membership re-entry, reconnect ownership, extended/zero cooldowns, burst pacing, auxiliary readiness (including EOSE before IPC settles), native socket refusal/retry/deadline/cancellation, unread fetch/classification and real nest trigger/HTTP/file persistence. - Earlier isolated renderer mutations that started readiness before enqueue or removed auxiliary priority each failed the real presence regression; restored controls passed. No native mutation claim. - Nest fixture uses explicit app-data/nest paths instead of assuming Windows known-folder APIs obey HOME/APPDATA. Windows runtime execution is not claimed; CI/build results remain separate. Originating conversation: buzz://message?channel=46ced308-71df-495d-a144-4ebf08334703&id=bc43c3422a3b0b2141268d0fa149b898421c824992b662aed8b73c4eee50404f No approval or merge recommendation is implied by opening this PR. --------- Signed-off-by: Brain <1a02c72794dcd0f07058a353bc3a81f4028b8c77c92c87fce6d5c8b85970a20b@buzz.block.builderlab.xyz> Co-authored-by: Brain <1a02c72794dcd0f07058a353bc3a81f4028b8c77c92c87fce6d5c8b85970a20b@buzz.block.builderlab.xyz>
This commit is contained in:
Wes
@@ -653,82 +653,84 @@ pub fn upsert_managed_section(file_path: &Path, new_section_content: &str) -> io
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Serializes nest-context writes so a slow, stale regeneration cannot roll the
|
||||
/// file back over a newer one. This is an ordered, latest-request-wins gate —
|
||||
/// not a work coalescer: every superseded generation still performs its relay
|
||||
/// reads, then drops its result at commit time. Adding a true dirty-loop owner
|
||||
/// would be a larger change and is unwarranted at this user-driven trigger rate.
|
||||
/// One regeneration worker with a latest-request-wins write fence. Startup
|
||||
/// persona backfill can request hundreds of renders: intermediate requests must
|
||||
/// supersede stale writes without each doing their own archive snapshot read.
|
||||
///
|
||||
/// Each regeneration request claims a monotonic generation *synchronously* at
|
||||
/// request time (see [`NestRegenGate::claim`]), so the generation encodes
|
||||
/// program order: boot's regen is claimed before `apply_workspace`'s, an edit's
|
||||
/// regen before the next edit's. The claimed generation travels with the
|
||||
/// spawned task and gates its write in [`NestRegenGate::commit`]: a task drops
|
||||
/// its result once a *newer generation has been requested*, even if that newer
|
||||
/// generation later fails before it writes. Gating on the highest *requested*
|
||||
/// generation — not the highest *written* one — is what stops a slow, stale
|
||||
/// pre-edit render from publishing after a newer post-edit render was claimed
|
||||
/// and then failed during its relay work (which would otherwise leave the
|
||||
/// obsolete roster authoritative until the next unrelated trigger). Declared
|
||||
/// semantic: once a newer regeneration is requested, no older one publishes;
|
||||
/// if that newer one fails, the file simply waits for the next trigger.
|
||||
///
|
||||
/// `claim` and `commit` share one lock, so the "is this still the newest
|
||||
/// request?" compare is atomic with the synchronous file write. A bare atomic
|
||||
/// watermark checked separately from the write would let a new claim slip
|
||||
/// between an older task's eligibility check and its write; holding the lock
|
||||
/// across both closes that window (no `await` occurs while it is held).
|
||||
/// Claiming, finishing and committing share one lock. A trigger during a read
|
||||
/// leaves one latest-generation follow-up; a trigger at worker shutdown either
|
||||
/// becomes that follow-up or starts a new worker. No debounce or cached archive
|
||||
/// state is needed. Once a newer generation is requested, an older one cannot
|
||||
/// publish, even if the newer render fails (the next trigger can try again).
|
||||
struct NestRegenGate {
|
||||
/// Highest generation *requested* so far (`0` = none yet). Advanced by
|
||||
/// [`claim`] and read by [`commit`]; guarding both under this single lock
|
||||
/// keeps the eligibility compare atomic with the file write.
|
||||
highest_requested: Mutex<u64>,
|
||||
state: Mutex<NestRegenState>,
|
||||
}
|
||||
|
||||
struct NestRegenState {
|
||||
highest_requested: u64,
|
||||
running: bool,
|
||||
}
|
||||
|
||||
impl NestRegenGate {
|
||||
const fn new() -> Self {
|
||||
Self {
|
||||
highest_requested: Mutex::new(0),
|
||||
state: Mutex::new(NestRegenState {
|
||||
highest_requested: 0,
|
||||
running: false,
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
/// Claim the next generation. Call synchronously at request time so the
|
||||
/// value reflects when the regeneration was requested, not when its task
|
||||
/// happens to run. Advancing the shared watermark here is what lets a later
|
||||
/// [`commit`] recognize — and drop — any older generation's stale render.
|
||||
fn claim(&self) -> u64 {
|
||||
let mut requested = self
|
||||
.highest_requested
|
||||
.lock()
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner());
|
||||
*requested += 1;
|
||||
*requested
|
||||
/// Claim synchronously, before spawning. Only the idle-to-running caller
|
||||
/// owns a worker; all other callers just advance the pending generation.
|
||||
fn claim(&self) -> (u64, bool) {
|
||||
let mut state = self.state.lock().unwrap_or_else(|p| p.into_inner());
|
||||
state.highest_requested += 1;
|
||||
let start_worker = !state.running;
|
||||
state.running = true;
|
||||
(state.highest_requested, start_worker)
|
||||
}
|
||||
|
||||
/// Non-blocking [`claim`] against the *exact* lock `claim` takes. Returns
|
||||
/// `Some(generation)` if it acquired the lock — i.e. a claim could proceed
|
||||
/// with no contention — or `None` if the lock is already held, meaning a
|
||||
/// concurrent claim would block on it. Because `claim` and `commit` share
|
||||
/// `highest_requested`, calling this from inside `commit_hooked`'s
|
||||
/// under-lock hook reports `None`: the eligibility compare and the write
|
||||
/// are serialized against any new claim. A design that advanced the
|
||||
/// watermark under a separate lock (or a lock-free atomic) would report
|
||||
/// `Some` here — the regression this probe proves absent, with no reliance
|
||||
/// on elapsed time or thread scheduling.
|
||||
/// Return work only to the caller that starts the worker. The callback is
|
||||
/// the real regeneration path, supplied here so tests can hold its I/O.
|
||||
fn request<'a, F, Fut>(
|
||||
&'a self,
|
||||
mut regenerate: F,
|
||||
) -> Option<impl std::future::Future<Output = ()> + 'a>
|
||||
where
|
||||
F: FnMut(u64) -> Fut + 'a,
|
||||
Fut: std::future::Future<Output = Result<(), String>> + 'a,
|
||||
{
|
||||
let (mut generation, start_worker) = self.claim();
|
||||
if !start_worker {
|
||||
return None;
|
||||
}
|
||||
Some(async move {
|
||||
loop {
|
||||
if let Err(error) = regenerate(generation).await {
|
||||
eprintln!("buzz-desktop: nest context regeneration failed: {error}");
|
||||
}
|
||||
let mut state = self.state.lock().unwrap_or_else(|p| p.into_inner());
|
||||
if state.highest_requested == generation {
|
||||
state.running = false;
|
||||
return;
|
||||
}
|
||||
generation = state.highest_requested;
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/// Probe the exact claim/commit lock, including inside the commit hook.
|
||||
#[cfg(test)]
|
||||
fn try_claim(&self) -> Option<u64> {
|
||||
match self.highest_requested.try_lock() {
|
||||
Ok(mut requested) => {
|
||||
*requested += 1;
|
||||
Some(*requested)
|
||||
}
|
||||
Err(std::sync::TryLockError::WouldBlock) => None,
|
||||
Err(std::sync::TryLockError::Poisoned(poisoned)) => {
|
||||
let mut requested = poisoned.into_inner();
|
||||
*requested += 1;
|
||||
Some(*requested)
|
||||
}
|
||||
}
|
||||
let mut state = match self.state.try_lock() {
|
||||
Ok(state) => state,
|
||||
Err(std::sync::TryLockError::WouldBlock) => return None,
|
||||
Err(std::sync::TryLockError::Poisoned(poisoned)) => poisoned.into_inner(),
|
||||
};
|
||||
state.highest_requested += 1;
|
||||
state.running = true;
|
||||
Some(state.highest_requested)
|
||||
}
|
||||
|
||||
/// Commit `content` for `generation`, dropping the write once a newer
|
||||
@@ -755,10 +757,10 @@ impl NestRegenGate {
|
||||
under_lock: impl FnOnce(),
|
||||
) -> io::Result<bool> {
|
||||
let requested = self
|
||||
.highest_requested
|
||||
.state
|
||||
.lock()
|
||||
.map_err(|_| io::Error::other("nest regen gate lock poisoned"))?;
|
||||
if generation < *requested {
|
||||
if generation < requested.highest_requested {
|
||||
return Ok(false);
|
||||
}
|
||||
under_lock();
|
||||
@@ -767,7 +769,12 @@ impl NestRegenGate {
|
||||
}
|
||||
}
|
||||
|
||||
/// Process-wide ordered write gate for nest-context regeneration.
|
||||
// A best-effort roster refresh must not strand every newer edit behind an old
|
||||
// relay's unbounded NIP-11 body or admission wait. Bound the complete archive
|
||||
// operation, not just request headers; timeout preserves the existing fail-open.
|
||||
const NEST_ARCHIVE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
|
||||
|
||||
/// Process-wide regeneration owner and ordered write gate.
|
||||
static NEST_REGEN: NestRegenGate = NestRegenGate::new();
|
||||
|
||||
pub async fn regenerate_nest_context<R: tauri::Runtime>(
|
||||
@@ -795,10 +802,20 @@ pub async fn regenerate_nest_context<R: tauri::Runtime>(
|
||||
// uses the same captured target as the rendered relay; a later generation's
|
||||
// task always wins the commit, so a fallback-relay boot render cannot bury a
|
||||
// later apply_workspace render.
|
||||
let archived: HashSet<String> = fetch_archived_pubkeys_at(&state, &target)
|
||||
.await
|
||||
.into_iter()
|
||||
.collect();
|
||||
let archived: HashSet<String> = match tokio::time::timeout(
|
||||
NEST_ARCHIVE_TIMEOUT,
|
||||
fetch_archived_pubkeys_at(&state, &target),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(pubkeys) => pubkeys.into_iter().collect(),
|
||||
Err(_) => {
|
||||
eprintln!(
|
||||
"buzz-desktop: nest archive read timed out; rendering without archive filter"
|
||||
);
|
||||
HashSet::new()
|
||||
}
|
||||
};
|
||||
let content = render_dynamic_section(&personas, &agents, &archived, &target.ws_url);
|
||||
NEST_REGEN
|
||||
.commit(&agents_md, &content, generation)
|
||||
@@ -807,27 +824,25 @@ pub async fn regenerate_nest_context<R: tauri::Runtime>(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Convenience wrapper: claims a regeneration generation, then regenerates on a
|
||||
/// spawned task, logging a warning on failure.
|
||||
///
|
||||
/// All call sites treat regeneration as fire-and-forget — agents run fine with
|
||||
/// a stale AGENTS.md, so we warn and continue rather than propagating the error.
|
||||
/// The generation is claimed *here*, synchronously, so it encodes call order;
|
||||
/// the spawned task carries it into [`NestRegenGate::commit`], which drops
|
||||
/// a stale render rather than letting a slow task overwrite a newer file.
|
||||
/// Archive/unarchive trigger this directly, but the regen races the relay's
|
||||
/// `kind:13535` snapshot update, so a just-archived agent may still linger for
|
||||
/// one cycle until the next regen (any agent/team edit or the next launch).
|
||||
/// Fire-and-forget regeneration: one worker reads the latest state, with one
|
||||
/// pending follow-up if another trigger arrives. Failures still warn and leave
|
||||
/// the file for the next trigger; they never strand the worker as running.
|
||||
/// Archive/unarchive can race the relay's snapshot update, so an archived agent
|
||||
/// may still linger until the next trigger, as before.
|
||||
pub fn try_regenerate_nest<R: tauri::Runtime>(app: &AppHandle<R>) {
|
||||
let generation = NEST_REGEN.claim();
|
||||
let app = app.clone();
|
||||
tauri::async_runtime::spawn(async move {
|
||||
if let Err(error) = regenerate_nest_context(&app, generation).await {
|
||||
eprintln!("buzz-desktop: nest context regeneration failed: {error}");
|
||||
}
|
||||
});
|
||||
if let Some(work) = NEST_REGEN.request(move |generation| {
|
||||
let app = app.clone();
|
||||
async move { regenerate_nest_context(&app, generation).await }
|
||||
}) {
|
||||
tauri::async_runtime::spawn(work);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod regen_tests;
|
||||
#[cfg(test)]
|
||||
mod regen_trigger_tests;
|
||||
#[cfg(test)]
|
||||
mod render_tests;
|
||||
#[cfg(test)]
|
||||
|
||||
120
desktop/src-tauri/src/managed_agents/nest/regen_tests.rs
Normal file
120
desktop/src-tauri/src/managed_agents/nest/regen_tests.rs
Normal file
@@ -0,0 +1,120 @@
|
||||
//! Exercise the production request/worker/commit boundary with held I/O and a
|
||||
//! real managed-section file. No sleeps, live relay, Tauri process or home writes.
|
||||
use super::*;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use tokio::sync::{mpsc, oneshot};
|
||||
|
||||
#[tokio::test]
|
||||
async fn burst_coalesces_reads_and_commits_only_the_latest_workspace() {
|
||||
let gate = NestRegenGate::new();
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let file = tmp.path().join("AGENTS.md");
|
||||
fs::write(&file, "# User instructions\n").unwrap();
|
||||
let current = Mutex::new("wss://before.example");
|
||||
let reads = AtomicUsize::new(0);
|
||||
let active = AtomicUsize::new(0);
|
||||
let (started, mut starts) = mpsc::unbounded_channel();
|
||||
let worker = gate
|
||||
.request(|generation| {
|
||||
let relay = *current.lock().unwrap();
|
||||
let (release, released) = oneshot::channel();
|
||||
reads.fetch_add(1, Ordering::SeqCst);
|
||||
assert_eq!(active.fetch_add(1, Ordering::SeqCst), 0);
|
||||
started.send((generation, release)).unwrap();
|
||||
let gate = &gate;
|
||||
let active = &active;
|
||||
let file = &file;
|
||||
async move {
|
||||
released.await.unwrap();
|
||||
let content = render_dynamic_section(&[], &[], &HashSet::new(), relay);
|
||||
gate.commit(file, &content, generation).unwrap();
|
||||
active.fetch_sub(1, Ordering::SeqCst);
|
||||
Ok(())
|
||||
}
|
||||
})
|
||||
.unwrap();
|
||||
tokio::pin!(worker);
|
||||
let (first, release_first) = tokio::select! {
|
||||
value = starts.recv() => value.unwrap(),
|
||||
() = &mut worker => panic!("worker completed before its read"),
|
||||
};
|
||||
assert_eq!(first, 1);
|
||||
|
||||
// Model startup backfill and a workspace switch while the first snapshot
|
||||
// read is held. New requests must not even start their expensive callback.
|
||||
*current.lock().unwrap() = "wss://latest.example";
|
||||
for _ in 0..289 {
|
||||
assert!(gate
|
||||
.request(|_| async { panic!("duplicate worker") })
|
||||
.is_none());
|
||||
}
|
||||
assert_eq!(reads.load(Ordering::SeqCst), 1);
|
||||
release_first.send(()).unwrap();
|
||||
let (latest, release_latest) = tokio::select! {
|
||||
value = starts.recv() => value.unwrap(),
|
||||
() = &mut worker => panic!("latest request was lost"),
|
||||
};
|
||||
assert_eq!(latest, 290);
|
||||
assert_eq!(reads.load(Ordering::SeqCst), 2);
|
||||
assert_eq!(fs::read_to_string(&file).unwrap(), "# User instructions\n");
|
||||
release_latest.send(()).unwrap();
|
||||
worker.await;
|
||||
|
||||
let content = fs::read_to_string(&file).unwrap();
|
||||
assert!(content.starts_with("# User instructions"));
|
||||
assert!(content.contains("wss://latest.example"));
|
||||
assert!(!content.contains("wss://before.example"));
|
||||
assert_eq!(active.load(Ordering::SeqCst), 0);
|
||||
assert!(!gate.state.lock().unwrap().running);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn trigger_during_follow_up_and_after_idle_is_not_lost() {
|
||||
let gate = NestRegenGate::new();
|
||||
let passes = AtomicUsize::new(0);
|
||||
gate.request(|generation| {
|
||||
let pass = passes.fetch_add(1, Ordering::SeqCst);
|
||||
if pass < 2 {
|
||||
assert!(gate
|
||||
.request(|_| async { panic!("second worker") })
|
||||
.is_none());
|
||||
}
|
||||
async move {
|
||||
assert_eq!(generation, pass as u64 + 1);
|
||||
Ok(())
|
||||
}
|
||||
})
|
||||
.unwrap()
|
||||
.await;
|
||||
assert_eq!(passes.load(Ordering::SeqCst), 3);
|
||||
assert!(!gate.state.lock().unwrap().running);
|
||||
|
||||
// A request after idle must acquire ownership, not remain stranded dirty.
|
||||
gate.request(|generation| async move {
|
||||
assert_eq!(generation, 4);
|
||||
Ok(())
|
||||
})
|
||||
.unwrap()
|
||||
.await;
|
||||
assert!(!gate.state.lock().unwrap().running);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn failed_pass_runs_pending_latest_and_releases_owner_for_next_trigger() {
|
||||
let gate = NestRegenGate::new();
|
||||
let passes = AtomicUsize::new(0);
|
||||
gate.request(|_| {
|
||||
if passes.fetch_add(1, Ordering::SeqCst) == 0 {
|
||||
assert!(gate
|
||||
.request(|_| async { panic!("second worker") })
|
||||
.is_none());
|
||||
}
|
||||
async { Err("controlled regeneration failure".into()) }
|
||||
})
|
||||
.unwrap()
|
||||
.await;
|
||||
assert_eq!(passes.load(Ordering::SeqCst), 2);
|
||||
assert!(!gate.state.lock().unwrap().running);
|
||||
gate.request(|_| async { Ok(()) }).unwrap().await;
|
||||
assert!(!gate.state.lock().unwrap().running);
|
||||
}
|
||||
170
desktop/src-tauri/src/managed_agents/nest/regen_trigger_tests.rs
Normal file
170
desktop/src-tauri/src/managed_agents/nest/regen_trigger_tests.rs
Normal file
@@ -0,0 +1,170 @@
|
||||
//! Real fire-and-forget trigger, real HTTP bodies, and real roster persistence.
|
||||
//! Run in a child process so the home directory, Tauri runtime and global nest
|
||||
//! owner cannot touch the user's profile or race unrelated package tests.
|
||||
use super::*;
|
||||
use std::sync::{
|
||||
atomic::{AtomicUsize, Ordering},
|
||||
Arc,
|
||||
};
|
||||
use std::time::Duration;
|
||||
|
||||
#[test]
|
||||
fn real_trigger_coalesces_and_survives_a_stalled_old_workspace() {
|
||||
const CHILD: &str = "BUZZ_NEST_REGEN_TEST_HOME";
|
||||
if let Some(home) = std::env::var_os(CHILD) {
|
||||
let home = PathBuf::from(home);
|
||||
assert_eq!(std::env::var_os("HOME").unwrap(), home.as_os_str());
|
||||
let runtime = tokio::runtime::Builder::new_current_thread()
|
||||
.enable_all()
|
||||
.build()
|
||||
.unwrap();
|
||||
runtime.block_on(async {
|
||||
tauri::async_runtime::set(tokio::runtime::Handle::current());
|
||||
exercise_real_trigger(&home).await;
|
||||
});
|
||||
return;
|
||||
}
|
||||
let home = tempfile::tempdir().unwrap();
|
||||
let output = std::process::Command::new(std::env::current_exe().unwrap())
|
||||
.args(["--exact", "managed_agents::nest::regen_trigger_tests::real_trigger_coalesces_and_survives_a_stalled_old_workspace", "--nocapture"])
|
||||
.env(CHILD, home.path()).env("HOME", home.path())
|
||||
.env("XDG_DATA_HOME", home.path()).env("APPDATA", home.path())
|
||||
.env("LOCALAPPDATA", home.path())
|
||||
.env_remove("BUZZ_PRIVATE_KEY").env_remove("BUZZ_AUTH_TAG")
|
||||
.env_remove("BUZZ_RELAY_URL").env_remove("BUZZ_NETWORK_TRACE")
|
||||
.output().unwrap();
|
||||
assert!(
|
||||
output.status.success(),
|
||||
"child failed: {}\n{}",
|
||||
String::from_utf8_lossy(&output.stdout),
|
||||
String::from_utf8_lossy(&output.stderr)
|
||||
);
|
||||
}
|
||||
|
||||
async fn exercise_real_trigger(home: &Path) {
|
||||
use axum::{
|
||||
body::{Body, Bytes},
|
||||
response::Response,
|
||||
routing::{get, post},
|
||||
Json, Router,
|
||||
};
|
||||
let old_hits = Arc::new(AtomicUsize::new(0));
|
||||
let old_started = Arc::new(tokio::sync::Notify::new());
|
||||
let hits = old_hits.clone();
|
||||
let started = old_started.clone();
|
||||
let old = Router::new().route(
|
||||
"/",
|
||||
get(move || {
|
||||
hits.fetch_add(1, Ordering::SeqCst);
|
||||
started.notify_one();
|
||||
async {
|
||||
// Headers succeed but the NIP-11 body never ends. Only the real
|
||||
// nest-owned timeout can release this read; the test never does.
|
||||
Response::new(Body::from_stream(futures_util::stream::pending::<
|
||||
Result<Bytes, std::io::Error>,
|
||||
>()))
|
||||
}
|
||||
}),
|
||||
);
|
||||
let keys = nostr::Keys::generate();
|
||||
let relay_self = keys.public_key().to_hex();
|
||||
let snapshot = nostr::EventBuilder::new(nostr::Kind::Custom(13535), "")
|
||||
.sign_with_keys(&keys)
|
||||
.unwrap();
|
||||
let query_hits = Arc::new(AtomicUsize::new(0));
|
||||
let hits = query_hits.clone();
|
||||
let latest = Router::new()
|
||||
.route(
|
||||
"/",
|
||||
get(move || {
|
||||
let value = relay_self.clone();
|
||||
async move { Json(serde_json::json!({"self": value})) }
|
||||
}),
|
||||
)
|
||||
.route(
|
||||
"/query",
|
||||
post(move || {
|
||||
hits.fetch_add(1, Ordering::SeqCst);
|
||||
let value = snapshot.clone();
|
||||
async move { Json(serde_json::json!([value])) }
|
||||
}),
|
||||
);
|
||||
async fn serve(router: Router) -> (String, tokio::task::JoinHandle<()>) {
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let url = format!("ws://{}", listener.local_addr().unwrap());
|
||||
(
|
||||
url,
|
||||
tokio::spawn(async move {
|
||||
axum::serve(listener, router).await.unwrap();
|
||||
}),
|
||||
)
|
||||
}
|
||||
let (old_url, old_server) = serve(old).await;
|
||||
let (latest_url, latest_server) = serve(latest).await;
|
||||
// Inject paths rather than relying on HOME/APPDATA: Windows known-folder
|
||||
// APIs ignore those environment overrides. PathResolver joins the mock
|
||||
// identifier onto data_dir; an absolute identifier replaces that base.
|
||||
let app_data = home.join("app-data");
|
||||
let mut context = tauri::test::mock_context(tauri::test::noop_assets());
|
||||
context.config_mut().identifier = app_data.to_str().unwrap().to_owned();
|
||||
NEST_DIR.set(Some(home.join("nest"))).unwrap();
|
||||
let state = crate::app_state::build_app_state();
|
||||
*state.relay_url_override.lock().unwrap() = Some(old_url.clone());
|
||||
let app = tauri::test::mock_builder()
|
||||
.manage(state)
|
||||
.build(context)
|
||||
.unwrap();
|
||||
assert_eq!(app.path().app_data_dir().unwrap(), app_data);
|
||||
let nest = nest_dir().unwrap();
|
||||
assert_eq!(nest, home.join("nest"));
|
||||
fs::create_dir_all(&nest).unwrap();
|
||||
let file = nest.join("AGENTS.md");
|
||||
fs::write(&file, "# User instructions\n").unwrap();
|
||||
|
||||
try_regenerate_nest(app.handle());
|
||||
tokio::time::timeout(Duration::from_secs(3), old_started.notified())
|
||||
.await
|
||||
.unwrap();
|
||||
for _ in 0..288 {
|
||||
try_regenerate_nest(app.handle());
|
||||
}
|
||||
*app.state::<AppState>().relay_url_override.lock().unwrap() = Some(latest_url.clone());
|
||||
try_regenerate_nest(app.handle());
|
||||
|
||||
// Wait through the actual operation deadline. Do not release the stalled
|
||||
// response or manufacture a fresh worker; its pending latest trigger must
|
||||
// progress on its own and finish with exactly one new-workspace query.
|
||||
tokio::time::timeout(NEST_ARCHIVE_TIMEOUT + Duration::from_secs(5), async {
|
||||
loop {
|
||||
let content = fs::read_to_string(&file).unwrap();
|
||||
assert!(!content.contains(&old_url), "stale workspace was committed");
|
||||
if content.contains(&latest_url) && !NEST_REGEN.state.lock().unwrap().running {
|
||||
break;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(10)).await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(old_hits.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(query_hits.load(Ordering::SeqCst), 1);
|
||||
assert!(fs::read_to_string(&file)
|
||||
.unwrap()
|
||||
.starts_with("# User instructions"));
|
||||
|
||||
// Idle-to-running through the real trigger must still work.
|
||||
try_regenerate_nest(app.handle());
|
||||
tokio::time::timeout(Duration::from_secs(3), async {
|
||||
loop {
|
||||
if !NEST_REGEN.state.lock().unwrap().running {
|
||||
break;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(10)).await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(query_hits.load(Ordering::SeqCst), 2);
|
||||
old_server.abort();
|
||||
latest_server.abort();
|
||||
}
|
||||
@@ -528,8 +528,8 @@ fn commit_newer_generation_wins_over_a_stale_finisher() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let file = agents_md_with_markers(tmp.path());
|
||||
|
||||
let gen_a = gate.claim(); // pre-edit request
|
||||
let gen_b = gate.claim(); // post-edit request
|
||||
let gen_a = gate.claim().0; // pre-edit request
|
||||
let gen_b = gate.claim().0; // post-edit request
|
||||
assert!(gen_a < gen_b);
|
||||
|
||||
// B (newer) commits first.
|
||||
@@ -558,8 +558,8 @@ fn commit_boot_fallback_relay_cannot_bury_apply_workspace_relay() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let file = agents_md_with_markers(tmp.path());
|
||||
|
||||
let boot_gen = gate.claim(); // boot, fallback relay
|
||||
let apply_gen = gate.claim(); // apply_workspace, workspace relay
|
||||
let boot_gen = gate.claim().0; // boot, fallback relay
|
||||
let apply_gen = gate.claim().0; // apply_workspace, workspace relay
|
||||
|
||||
// apply_workspace's render lands first.
|
||||
assert!(gate
|
||||
@@ -599,8 +599,8 @@ fn commit_failed_newer_request_still_supersedes_older_snapshot() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let file = agents_md_with_markers(tmp.path());
|
||||
|
||||
let gen1 = gate.claim(); // pre-edit request
|
||||
let gen2 = gate.claim(); // post-edit request
|
||||
let gen1 = gate.claim().0; // pre-edit request
|
||||
let gen2 = gate.claim().0; // post-edit request
|
||||
assert!(gen1 < gen2);
|
||||
|
||||
// gen2 fails during relay work and never reaches commit — nothing written.
|
||||
@@ -640,7 +640,7 @@ fn commit_claim_at_the_older_tasks_cutover_supersedes_it() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let file = agents_md_with_markers(tmp.path());
|
||||
|
||||
let gen1 = gate.claim();
|
||||
let gen1 = gate.claim().0;
|
||||
|
||||
let wrote_gen1 = gate
|
||||
.commit_hooked(&file, "gen1 roster", gen1, || {
|
||||
@@ -664,7 +664,7 @@ fn commit_claim_at_the_older_tasks_cutover_supersedes_it() {
|
||||
|
||||
// The lock is free once commit returns, so a newer request now claims and
|
||||
// may publish over gen1.
|
||||
let gen2 = gate.claim();
|
||||
let gen2 = gate.claim().0;
|
||||
assert!(gen1 < gen2);
|
||||
assert!(gate.commit(&file, "gen2 roster", gen2).unwrap());
|
||||
|
||||
@@ -681,7 +681,7 @@ fn commit_equal_generation_is_allowed() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let file = agents_md_with_markers(tmp.path());
|
||||
|
||||
let gen = gate.claim();
|
||||
let gen = gate.claim().0;
|
||||
assert!(gate.commit(&file, "first", gen).unwrap());
|
||||
assert!(
|
||||
gate.commit(&file, "second", gen).unwrap(),
|
||||
@@ -702,11 +702,11 @@ fn commit_poisoned_lock_returns_error_instead_of_panicking() {
|
||||
let gate = std::sync::Arc::new(NestRegenGate::new());
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let file = agents_md_with_markers(tmp.path());
|
||||
let gen = gate.claim();
|
||||
let gen = gate.claim().0;
|
||||
|
||||
let poisoner = gate.clone();
|
||||
let _ = std::thread::spawn(move || {
|
||||
let _guard = poisoner.highest_requested.lock().unwrap();
|
||||
let _guard = poisoner.state.lock().unwrap();
|
||||
panic!("poison the gate lock");
|
||||
})
|
||||
.join();
|
||||
|
||||
@@ -239,7 +239,30 @@ pub(crate) struct RelaySession {
|
||||
|
||||
struct PendingRequest {
|
||||
events: Vec<Event>,
|
||||
seen: HashSet<nostr::EventId>,
|
||||
complete: oneshot::Sender<Result<Vec<Event>, String>>,
|
||||
deadline: Instant,
|
||||
retry: ClosedRetry,
|
||||
}
|
||||
|
||||
impl PendingRequest {
|
||||
fn is_active(&self) -> bool {
|
||||
!self.complete.is_closed() && Instant::now() < self.deadline
|
||||
}
|
||||
}
|
||||
|
||||
// Close before waking, including when fetch_events is dropped during registration.
|
||||
// The existing session owner reclaims state; Drop does not spawn another owner.
|
||||
struct RequestReceiver<'a> {
|
||||
result: oneshot::Receiver<Result<Vec<Event>, String>>,
|
||||
wake: &'a mpsc::Sender<()>,
|
||||
}
|
||||
|
||||
impl Drop for RequestReceiver<'_> {
|
||||
fn drop(&mut self) {
|
||||
self.result.close();
|
||||
let _ = self.wake.try_send(());
|
||||
}
|
||||
}
|
||||
|
||||
/// Desired set plus the write-time record of what has left it.
|
||||
@@ -304,12 +327,20 @@ impl RelaySession {
|
||||
timeout: Duration,
|
||||
) -> Result<Vec<Event>, String> {
|
||||
let id = format!("native-fetch-{}", uuid::Uuid::new_v4());
|
||||
let deadline = Instant::now() + timeout;
|
||||
let (complete, result) = oneshot::channel();
|
||||
let mut receiver = RequestReceiver {
|
||||
result,
|
||||
wake: &self.wake,
|
||||
};
|
||||
self.requests.lock().await.insert(
|
||||
id.clone(),
|
||||
PendingRequest {
|
||||
events: Vec::new(),
|
||||
seen: HashSet::new(),
|
||||
complete,
|
||||
deadline,
|
||||
retry: ClosedRetry::default(),
|
||||
},
|
||||
);
|
||||
{
|
||||
@@ -323,7 +354,7 @@ impl RelaySession {
|
||||
|
||||
let outcome = tokio::select! {
|
||||
_ = self.cancel.cancelled() => Err("relay session cancelled".to_string()),
|
||||
value = tokio::time::timeout(timeout, result) => match value {
|
||||
value = tokio::time::timeout_at(deadline, &mut receiver.result) => match value {
|
||||
Ok(Ok(value)) => value,
|
||||
Ok(Err(_)) => Err("relay request ended before EOSE".to_string()),
|
||||
Err(_) => Err("relay request timed out".to_string()),
|
||||
@@ -334,14 +365,33 @@ impl RelaySession {
|
||||
}
|
||||
|
||||
async fn finish_request(&self, id: &str) {
|
||||
self.requests.lock().await.remove(id);
|
||||
let mut requests = self.requests.lock().await;
|
||||
let mut state = self.state.lock().await;
|
||||
// Do not leave an orphan transient if the caller drops during cleanup.
|
||||
requests.remove(id);
|
||||
state.transient.retain(|subscription| subscription.id != id);
|
||||
state.removed.insert(id.to_string());
|
||||
drop(state);
|
||||
let _ = self.wake.try_send(());
|
||||
}
|
||||
|
||||
// Also called while connecting/backing off: caller cancellation must reclaim
|
||||
// buffers even if authentication never succeeds and reconcile cannot run.
|
||||
async fn prune_cancelled_requests(&self) {
|
||||
let mut requests = self.requests.lock().await;
|
||||
let mut removed = Vec::new();
|
||||
requests.retain(|id, request| {
|
||||
if !request.complete.is_closed() {
|
||||
return true;
|
||||
}
|
||||
removed.push(id.clone());
|
||||
false
|
||||
});
|
||||
let mut state = self.state.lock().await;
|
||||
state.transient.retain(|sub| !removed.contains(&sub.id));
|
||||
state.removed.extend(removed);
|
||||
}
|
||||
|
||||
/// Replaces the desired subscription set and wakes the loop to reconcile.
|
||||
///
|
||||
/// Reconciliation is declarative rather than incremental: callers state
|
||||
@@ -428,7 +478,17 @@ async fn run_session(
|
||||
return;
|
||||
}
|
||||
|
||||
match NostrWsConnection::connect_authenticated(&relay_url, &keys, auth_tag.as_ref()).await {
|
||||
let connecting =
|
||||
NostrWsConnection::connect_authenticated(&relay_url, &keys, auth_tag.as_ref());
|
||||
tokio::pin!(connecting);
|
||||
let connected = loop {
|
||||
tokio::select! {
|
||||
_ = session.cancel.cancelled() => return,
|
||||
Some(()) = wake_rx.recv() => session.prune_cancelled_requests().await,
|
||||
result = &mut connecting => break result,
|
||||
}
|
||||
};
|
||||
match connected {
|
||||
Ok(conn) => {
|
||||
// A connection that authenticated is healthy regardless of how
|
||||
// long it then lived, so backoff resets here rather than on
|
||||
@@ -445,9 +505,14 @@ async fn run_session(
|
||||
if session.cancel.is_cancelled() {
|
||||
return;
|
||||
}
|
||||
tokio::select! {
|
||||
_ = session.cancel.cancelled() => return,
|
||||
_ = tokio::time::sleep(delay) => {}
|
||||
let backoff = tokio::time::sleep(delay);
|
||||
tokio::pin!(backoff);
|
||||
loop {
|
||||
tokio::select! {
|
||||
_ = session.cancel.cancelled() => return,
|
||||
Some(()) = wake_rx.recv() => session.prune_cancelled_requests().await,
|
||||
_ = &mut backoff => break,
|
||||
}
|
||||
}
|
||||
delay = (delay * 2).min(RECONNECT_MAX_DELAY);
|
||||
}
|
||||
@@ -496,7 +561,19 @@ async fn run_connection(
|
||||
// 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();
|
||||
let finite_retry_at = session
|
||||
.requests
|
||||
.lock()
|
||||
.await
|
||||
.values()
|
||||
.filter(|request| request.is_active())
|
||||
.filter_map(|request| request.retry.due_at)
|
||||
.min();
|
||||
let retry_at = retries
|
||||
.values()
|
||||
.filter_map(|retry| retry.due_at)
|
||||
.chain(finite_retry_at)
|
||||
.min();
|
||||
|
||||
tokio::select! {
|
||||
_ = session.cancel.cancelled() => {
|
||||
@@ -515,6 +592,11 @@ async fn run_connection(
|
||||
_ = tokio::time::sleep_until(retry_at.unwrap_or_else(Instant::now)),
|
||||
if retry_at.is_some() =>
|
||||
{
|
||||
for request in session.requests.lock().await.values_mut() {
|
||||
if request.retry.due_at.is_some_and(|due| due <= Instant::now()) {
|
||||
request.retry.due_at = None;
|
||||
}
|
||||
}
|
||||
for retry in retries.values_mut() {
|
||||
if retry.due_at.is_some_and(|due| due <= Instant::now()) {
|
||||
retry.due_at = None;
|
||||
@@ -559,7 +641,9 @@ async fn run_connection(
|
||||
.await
|
||||
.get_mut(&subscription_id)
|
||||
{
|
||||
request.events.push(*event);
|
||||
if request.is_active() && request.seen.insert(event.id) {
|
||||
request.events.push(*event);
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
@@ -598,7 +682,17 @@ async fn run_connection(
|
||||
if open.remove(&subscription_id).is_none() {
|
||||
continue;
|
||||
}
|
||||
if let Some(request) = session.requests.lock().await.remove(&subscription_id) {
|
||||
let mut requests = session.requests.lock().await;
|
||||
if let Some(request) = requests.get_mut(&subscription_id) {
|
||||
// Finite quota recovery belongs to the request, not the
|
||||
// socket: reconnects and partial EVENTs cannot reset it.
|
||||
if request.is_active() && classify_closed(&message) == ClosedClass::RateLimited
|
||||
&& request.retry.attempts < 3 {
|
||||
request.retry.schedule(&message);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
if let Some(request) = requests.remove(&subscription_id) {
|
||||
let _ = request.complete.send(Err(format!("relay closed request: {message}")));
|
||||
let mut state = session.state.lock().await;
|
||||
state.transient.retain(|subscription| subscription.id != subscription_id);
|
||||
@@ -607,6 +701,7 @@ async fn run_connection(
|
||||
let _ = session.wake.try_send(());
|
||||
continue;
|
||||
}
|
||||
drop(requests);
|
||||
let retry = retries.entry(subscription_id.clone()).or_default();
|
||||
retry.schedule(&message);
|
||||
eprintln!(
|
||||
@@ -620,8 +715,12 @@ async fn run_connection(
|
||||
// is what keeps an intermittent relay from ratcheting
|
||||
// its way to the 30s ceiling and staying there.
|
||||
let was_open = open.contains_key(&subscription_id);
|
||||
if let Some(request) = session.requests.lock().await.remove(&subscription_id) {
|
||||
let _ = request.complete.send(Ok(request.events));
|
||||
let mut requests = session.requests.lock().await;
|
||||
if requests.contains_key(&subscription_id) && !was_open { continue; }
|
||||
if let Some(request) = requests.remove(&subscription_id) {
|
||||
let result = if request.is_active() { Ok(request.events) }
|
||||
else { Err("relay request timed out or cancelled".to_string()) };
|
||||
let _ = request.complete.send(result);
|
||||
let mut state = session.state.lock().await;
|
||||
state.transient.retain(|subscription| subscription.id != subscription_id);
|
||||
state.removed.insert(subscription_id.clone());
|
||||
@@ -629,6 +728,7 @@ async fn run_connection(
|
||||
let _ = session.wake.try_send(());
|
||||
continue;
|
||||
}
|
||||
drop(requests);
|
||||
retries.remove(&subscription_id);
|
||||
// The relay is running a subscription this socket does
|
||||
// not think is open, so the two disagree. EOSE is the
|
||||
@@ -676,6 +776,7 @@ async fn reconcile(
|
||||
// `set_subscriptions` land in the gap, spending its removal against a
|
||||
// desired set captured before it — reopening a subscription the caller had
|
||||
// just dropped, with no record left to catch it on the next pass.
|
||||
session.prune_cancelled_requests().await;
|
||||
let (desired, removed) = {
|
||||
let mut state = session.state.lock().await;
|
||||
let removed = std::mem::take(&mut state.removed);
|
||||
@@ -683,8 +784,9 @@ async fn reconcile(
|
||||
state
|
||||
.desired
|
||||
.iter()
|
||||
.chain(&state.transient)
|
||||
.cloned()
|
||||
.map(|sub| (sub, false))
|
||||
.chain(state.transient.iter().cloned().map(|sub| (sub, true)))
|
||||
.collect::<Vec<_>>(),
|
||||
removed,
|
||||
)
|
||||
@@ -699,7 +801,7 @@ async fn reconcile(
|
||||
}
|
||||
|
||||
for id in open.keys().cloned().collect::<Vec<_>>() {
|
||||
if desired.iter().any(|s| s.id == id) {
|
||||
if desired.iter().any(|(s, _)| s.id == id) {
|
||||
continue;
|
||||
}
|
||||
if conn
|
||||
@@ -712,7 +814,7 @@ async fn reconcile(
|
||||
open.remove(&id);
|
||||
}
|
||||
|
||||
for sub in desired {
|
||||
for (sub, finite) 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) {
|
||||
@@ -725,6 +827,22 @@ async fn reconcile(
|
||||
if retries.get(&sub.id).is_some_and(ClosedRetry::is_blocked) {
|
||||
continue;
|
||||
}
|
||||
if session.cancel.is_cancelled() {
|
||||
return false;
|
||||
}
|
||||
if finite {
|
||||
let requests = session.requests.lock().await;
|
||||
if !requests
|
||||
.get(&sub.id)
|
||||
.is_some_and(|request| request.is_active() && !request.retry.is_blocked())
|
||||
{
|
||||
continue;
|
||||
}
|
||||
}
|
||||
// The request-lock acquisition above can suspend through shutdown.
|
||||
if session.cancel.is_cancelled() {
|
||||
return false;
|
||||
}
|
||||
if conn
|
||||
.send_raw(&serde_json::json!(["REQ", sub.id, sub.filter]))
|
||||
.await
|
||||
@@ -786,7 +904,9 @@ impl ClosedRetry {
|
||||
.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.due_at = Instant::now().checked_add(self.backoff().max(hinted));
|
||||
// An unrepresentable hint is a hold, never an immediate retry.
|
||||
self.terminal = self.due_at.is_none();
|
||||
self.attempts = self.attempts.saturating_add(1);
|
||||
}
|
||||
ClosedClass::Retryable => {
|
||||
|
||||
459
desktop/src-tauri/src/native_relay_client_finite_tests.rs
Normal file
459
desktop/src-tauri/src/native_relay_client_finite_tests.rs
Normal file
@@ -0,0 +1,459 @@
|
||||
//! Finite CLOSED recovery through the real native session and WebSocket loop.
|
||||
use super::*;
|
||||
|
||||
fn fetch(
|
||||
session: &Arc<RelaySession>,
|
||||
timeout: Duration,
|
||||
) -> tokio::task::JoinHandle<Result<Vec<Event>, String>> {
|
||||
let session = Arc::clone(session);
|
||||
tokio::spawn(async move {
|
||||
session
|
||||
.fetch_events(
|
||||
serde_json::json!({"kinds": [9], "#h": ["unread"], "since": 42, "limit": 1000}),
|
||||
timeout,
|
||||
)
|
||||
.await
|
||||
})
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn quota_retry_preserves_events_without_duplicates_and_keeps_archive_live() {
|
||||
let (url, mut frames, commands) = stub_relay().await;
|
||||
let (session, mut archive) = start(url, Keys::generate(), None).await;
|
||||
session.set_subscriptions(vec![probe_subscription()]).await;
|
||||
assert_eq!(next_req(&mut frames, "archive").await, PROBE_ID);
|
||||
let pending = fetch(&session, Duration::from_secs(10));
|
||||
let id = next_req(&mut frames, "finite").await;
|
||||
let event = EventBuilder::text_note("retained")
|
||||
.sign_with_keys(&Keys::generate())
|
||||
.unwrap();
|
||||
let encoded = serde_json::to_value(&event).unwrap();
|
||||
commands
|
||||
.send(StubCommand::Event(id.clone(), encoded.clone()))
|
||||
.await
|
||||
.unwrap();
|
||||
let refused = Instant::now();
|
||||
commands
|
||||
.send(StubCommand::Closed(
|
||||
id.clone(),
|
||||
"rate-limited: quota exceeded; retry in 1s".into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
// Non-finite delivery is a receive-loop barrier, not a sleep or test-side retry setup.
|
||||
commands
|
||||
.send(StubCommand::Event(PROBE_ID.into(), encoded.clone()))
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
*tokio::time::timeout(Duration::from_secs(3), archive.recv())
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.event,
|
||||
event
|
||||
);
|
||||
assert!(
|
||||
!pending.is_finished(),
|
||||
"quota refusal must retain the finite operation"
|
||||
);
|
||||
assert_eq!(next_req(&mut frames, "quota retry").await, id);
|
||||
assert!(refused.elapsed() >= Duration::from_secs(1));
|
||||
commands
|
||||
.send(StubCommand::Event(id.clone(), encoded))
|
||||
.await
|
||||
.unwrap();
|
||||
commands.send(StubCommand::Eose(id.clone())).await.unwrap();
|
||||
assert_eq!(pending.await.unwrap().unwrap(), vec![event]);
|
||||
assert_eq!(
|
||||
next_frame(&mut frames, "finite CLOSE").await,
|
||||
Frame::Close(id)
|
||||
);
|
||||
session.shutdown();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn zero_hint_exhausts_three_retries_as_an_error() {
|
||||
let (url, mut frames, commands) = stub_relay().await;
|
||||
let (session, _archive) = start(url, Keys::generate(), None).await;
|
||||
let pending = fetch(&session, Duration::from_secs(12));
|
||||
let id = next_req(&mut frames, "finite").await;
|
||||
for attempt in 0..4 {
|
||||
commands
|
||||
.send(StubCommand::Closed(
|
||||
id.clone(),
|
||||
"rate-limited: quota exceeded; retry in 0s".into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
if attempt < 3 {
|
||||
assert_eq!(next_req(&mut frames, "bounded retry").await, id);
|
||||
}
|
||||
}
|
||||
let error = pending.await.unwrap().unwrap_err();
|
||||
assert!(error.contains("rate-limited:"), "{error}");
|
||||
assert!(session.requests.lock().await.is_empty());
|
||||
session.shutdown();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn terminal_refusal_and_deadline_do_not_become_empty_success() {
|
||||
for message in [
|
||||
"restricted: denied",
|
||||
"error: transient",
|
||||
"rate-limited: quota exceeded; retry in 10s",
|
||||
"rate-limited: quota exceeded",
|
||||
"rate-limited: quota exceeded; retry in 18446744073709551615s",
|
||||
] {
|
||||
let (url, mut frames, commands) = stub_relay().await;
|
||||
let (session, _archive) = start(url, Keys::generate(), None).await;
|
||||
let pending = fetch(&session, Duration::from_millis(300));
|
||||
let id = next_req(&mut frames, "finite").await;
|
||||
commands
|
||||
.send(StubCommand::Closed(id, message.into()))
|
||||
.await
|
||||
.unwrap();
|
||||
let error = pending.await.unwrap().unwrap_err();
|
||||
if message.starts_with("rate-limited:") {
|
||||
assert!(error.contains("timed out"), "{error}");
|
||||
} else {
|
||||
assert!(error.contains(message), "{error}");
|
||||
}
|
||||
assert!(session.requests.lock().await.is_empty());
|
||||
assert!(session.state.lock().await.transient.is_empty());
|
||||
assert!(
|
||||
frames.try_recv().is_err(),
|
||||
"must not dispatch through denial/deadline"
|
||||
);
|
||||
session.shutdown();
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn dropping_finite_caller_closes_request_without_stopping_archive() {
|
||||
let (url, mut frames, commands) = stub_relay().await;
|
||||
let (session, mut archive) = start(url, Keys::generate(), None).await;
|
||||
session.set_subscriptions(vec![probe_subscription()]).await;
|
||||
assert_eq!(next_req(&mut frames, "archive").await, PROBE_ID);
|
||||
let pending = fetch(&session, Duration::from_secs(10));
|
||||
let id = next_req(&mut frames, "finite").await;
|
||||
pending.abort();
|
||||
assert!(pending.await.unwrap_err().is_cancelled());
|
||||
assert_eq!(
|
||||
next_frame(&mut frames, "cancelled finite CLOSE").await,
|
||||
Frame::Close(id)
|
||||
);
|
||||
assert!(session.requests.lock().await.is_empty());
|
||||
assert!(session.state.lock().await.transient.is_empty());
|
||||
let event = EventBuilder::text_note("archive remains live")
|
||||
.sign_with_keys(&Keys::generate())
|
||||
.unwrap();
|
||||
commands
|
||||
.send(StubCommand::Event(
|
||||
PROBE_ID.into(),
|
||||
serde_json::to_value(&event).unwrap(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
*tokio::time::timeout(Duration::from_secs(3), archive.recv())
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.event,
|
||||
event
|
||||
);
|
||||
session.shutdown();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn reconnect_preserves_quota_hold_and_partial_events_do_not_reset_budget() {
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let url = format!("ws://{}", listener.local_addr().unwrap());
|
||||
let (seen_tx, mut seen_rx) = mpsc::channel(8);
|
||||
let event = EventBuilder::text_note("partial replay")
|
||||
.sign_with_keys(&Keys::generate())
|
||||
.unwrap();
|
||||
let server = tokio::spawn(async move {
|
||||
let mut original: Option<serde_json::Value> = None;
|
||||
let mut refused = std::time::Instant::now();
|
||||
for connection in 0..2 {
|
||||
let (tcp, _) = listener.accept().await.unwrap();
|
||||
let mut ws = tokio_tungstenite::accept_async(tcp).await.unwrap();
|
||||
ws.send(Message::Text(
|
||||
serde_json::json!(["AUTH", "reconnect-test"])
|
||||
.to_string()
|
||||
.into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
let mut requests = 0;
|
||||
while let Some(Ok(Message::Text(text))) = ws.next().await {
|
||||
let frame: serde_json::Value = serde_json::from_str(&text).unwrap();
|
||||
match frame[0].as_str() {
|
||||
Some("AUTH") => ws
|
||||
.send(Message::Text(
|
||||
serde_json::json!(["OK", frame[1]["id"], true, ""])
|
||||
.to_string()
|
||||
.into(),
|
||||
))
|
||||
.await
|
||||
.unwrap(),
|
||||
Some("REQ") => {
|
||||
if let Some(expected) = &original {
|
||||
assert_eq!(&frame, expected);
|
||||
assert!(
|
||||
refused.elapsed()
|
||||
>= Duration::from_secs(if connection == 1 && requests == 0 {
|
||||
2
|
||||
} else {
|
||||
1
|
||||
}),
|
||||
"reconnect must not erase due time"
|
||||
);
|
||||
} else {
|
||||
original = Some(frame.clone());
|
||||
}
|
||||
seen_tx.send(frame.clone()).await.unwrap();
|
||||
ws.send(Message::Text(
|
||||
serde_json::json!(["EVENT", frame[1], event])
|
||||
.to_string()
|
||||
.into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
let hint = if connection == 0 { 2 } else { 0 };
|
||||
refused = std::time::Instant::now();
|
||||
ws.send(Message::Text(
|
||||
serde_json::json!([
|
||||
"CLOSED",
|
||||
frame[1],
|
||||
format!("rate-limited: quota exceeded; retry in {hint}s")
|
||||
])
|
||||
.to_string()
|
||||
.into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
requests += 1;
|
||||
if connection == 0 {
|
||||
ws.close(None).await.unwrap();
|
||||
break;
|
||||
}
|
||||
if requests == 3 {
|
||||
return;
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
let (session, _archive) = start(url, Keys::generate(), None).await;
|
||||
let pending = fetch(&session, Duration::from_secs(15));
|
||||
let error = tokio::time::timeout(Duration::from_secs(12), pending)
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
.unwrap_err();
|
||||
assert!(error.contains("rate-limited:"), "{error}");
|
||||
let mut count = 0;
|
||||
while seen_rx.try_recv().is_ok() {
|
||||
count += 1;
|
||||
}
|
||||
assert_eq!(
|
||||
count, 4,
|
||||
"initial request plus three retries across both sockets"
|
||||
);
|
||||
assert!(session.requests.lock().await.is_empty());
|
||||
server.await.unwrap();
|
||||
session.shutdown();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn caller_drop_during_authentication_is_reclaimed_without_restarting_connect() {
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let url = format!("ws://{}", listener.local_addr().unwrap());
|
||||
let (connected_tx, connected) = oneshot::channel();
|
||||
let (release, release_rx) = oneshot::channel();
|
||||
let server = tokio::spawn(async move {
|
||||
let (tcp, _) = listener.accept().await.unwrap();
|
||||
let mut ws = tokio_tungstenite::accept_async(tcp).await.unwrap();
|
||||
connected_tx.send(()).unwrap();
|
||||
release_rx.await.unwrap();
|
||||
ws.send(Message::Text(
|
||||
serde_json::json!(["AUTH", "held-auth"]).to_string().into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
let Some(Ok(Message::Text(text))) = ws.next().await else {
|
||||
panic!("same authentication attempt must survive wake")
|
||||
};
|
||||
let auth: serde_json::Value = serde_json::from_str(&text).unwrap();
|
||||
assert_eq!(auth[0], "AUTH");
|
||||
ws.send(Message::Text(
|
||||
serde_json::json!(["OK", auth[1]["id"], true, ""])
|
||||
.to_string()
|
||||
.into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
let next = tokio::time::timeout(Duration::from_millis(200), ws.next()).await;
|
||||
assert!(next.is_err(), "cancelled request must not be replayed");
|
||||
});
|
||||
let (session, _archive) = start(url, Keys::generate(), None).await;
|
||||
connected.await.unwrap();
|
||||
let pending = fetch(&session, Duration::from_secs(10));
|
||||
tokio::time::timeout(Duration::from_secs(2), async {
|
||||
loop {
|
||||
if !session.state.lock().await.transient.is_empty() {
|
||||
break;
|
||||
}
|
||||
tokio::task::yield_now().await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
pending.abort();
|
||||
assert!(pending.await.unwrap_err().is_cancelled());
|
||||
tokio::time::timeout(Duration::from_secs(2), async {
|
||||
loop {
|
||||
if session.requests.lock().await.is_empty()
|
||||
&& session.state.lock().await.transient.is_empty()
|
||||
{
|
||||
break;
|
||||
}
|
||||
tokio::task::yield_now().await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("cleanup must not require successful auth");
|
||||
release.send(()).unwrap();
|
||||
server.await.unwrap();
|
||||
session.shutdown();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cancellation_and_stale_eose_during_hold_cannot_complete_or_retry() {
|
||||
for shutdown in [false, true] {
|
||||
let (url, mut frames, commands) = stub_relay().await;
|
||||
let (session, mut archive) = start(url, Keys::generate(), None).await;
|
||||
session.set_subscriptions(vec![probe_subscription()]).await;
|
||||
assert_eq!(next_req(&mut frames, "archive").await, PROBE_ID);
|
||||
let pending = fetch(&session, Duration::from_secs(10));
|
||||
let id = next_req(&mut frames, "finite").await;
|
||||
commands
|
||||
.send(StubCommand::Closed(
|
||||
id.clone(),
|
||||
"rate-limited: quota exceeded; retry in 1s".into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
commands.send(StubCommand::Eose(id)).await.unwrap();
|
||||
let event = EventBuilder::text_note("barrier")
|
||||
.sign_with_keys(&Keys::generate())
|
||||
.unwrap();
|
||||
commands
|
||||
.send(StubCommand::Event(
|
||||
PROBE_ID.into(),
|
||||
serde_json::to_value(event).unwrap(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
tokio::time::timeout(Duration::from_secs(2), archive.recv())
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert!(
|
||||
!pending.is_finished(),
|
||||
"stale EOSE must not finish refused history"
|
||||
);
|
||||
if shutdown {
|
||||
session.shutdown();
|
||||
assert!(pending.await.unwrap().unwrap_err().contains("cancelled"));
|
||||
} else {
|
||||
pending.abort();
|
||||
assert!(pending.await.unwrap_err().is_cancelled());
|
||||
}
|
||||
tokio::time::timeout(Duration::from_secs(2), async {
|
||||
loop {
|
||||
if session.requests.lock().await.is_empty()
|
||||
&& session.state.lock().await.transient.is_empty()
|
||||
{
|
||||
break;
|
||||
}
|
||||
tokio::task::yield_now().await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(
|
||||
tokio::time::timeout(Duration::from_millis(1100), frames.recv())
|
||||
.await
|
||||
.ok()
|
||||
.flatten()
|
||||
.is_none(),
|
||||
"cancelled retry must never reopen"
|
||||
);
|
||||
session.shutdown();
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn timeout_cleanup_remains_owned_when_caller_drops_under_state_contention() {
|
||||
let (url, mut frames, _commands) = stub_relay().await;
|
||||
let (session, _archive) = start(url, Keys::generate(), None).await;
|
||||
let pending = fetch(&session, Duration::from_millis(150));
|
||||
let id = next_req(&mut frames, "finite").await;
|
||||
let state = session.state.lock().await;
|
||||
// Hold the real cleanup lock beyond the caller's absolute deadline. Abort
|
||||
// only after cleanup has been runnable, then let the existing loop reclaim.
|
||||
tokio::time::sleep(Duration::from_millis(250)).await;
|
||||
pending.abort();
|
||||
assert!(pending.await.unwrap_err().is_cancelled());
|
||||
drop(state);
|
||||
assert_eq!(
|
||||
next_frame(&mut frames, "cancelled cleanup CLOSE").await,
|
||||
Frame::Close(id)
|
||||
);
|
||||
assert!(session.requests.lock().await.is_empty());
|
||||
assert!(session.state.lock().await.transient.is_empty());
|
||||
session.shutdown();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cancellation_during_registration_does_not_leave_transient_state() {
|
||||
let (url, mut frames, _commands) = stub_relay().await;
|
||||
let (session, _archive) = start(url, Keys::generate(), None).await;
|
||||
session.set_subscriptions(vec![probe_subscription()]).await;
|
||||
assert_eq!(next_req(&mut frames, "archive").await, PROBE_ID);
|
||||
let state = session.state.lock().await;
|
||||
let pending = fetch(&session, Duration::from_secs(10));
|
||||
tokio::time::timeout(Duration::from_secs(2), async {
|
||||
loop {
|
||||
if !session.requests.lock().await.is_empty() {
|
||||
break;
|
||||
}
|
||||
tokio::task::yield_now().await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
pending.abort();
|
||||
assert!(pending.await.unwrap_err().is_cancelled());
|
||||
drop(state);
|
||||
tokio::time::timeout(Duration::from_secs(2), async {
|
||||
loop {
|
||||
if session.requests.lock().await.is_empty()
|
||||
&& session.state.lock().await.transient.is_empty()
|
||||
{
|
||||
break;
|
||||
}
|
||||
tokio::task::yield_now().await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(frames.try_recv().is_err());
|
||||
session.shutdown();
|
||||
}
|
||||
@@ -894,3 +894,6 @@ async fn the_first_lease_installs_a_session_the_archive_then_reuses() {
|
||||
|
||||
#[path = "native_relay_client_transport_tests.rs"]
|
||||
mod transport_tests;
|
||||
|
||||
#[path = "native_relay_client_finite_tests.rs"]
|
||||
mod finite_recovery_tests;
|
||||
|
||||
@@ -153,6 +153,33 @@ pub(crate) async fn unread_catch_up(
|
||||
// particular do not move the lease into a task or narrow its scope.
|
||||
let session = relay_client.session(relay_url.clone(), keys).await;
|
||||
|
||||
let (fetched, failures) = fetch_channels(session.handle(), &request).await?;
|
||||
|
||||
let current_keys = state.signing_keys()?;
|
||||
if current_keys.public_key().to_hex() != owner
|
||||
|| crate::relay::relay_ws_url_with_override(&state) != relay_url
|
||||
{
|
||||
return Err("unread catch-up scope changed while fetching".to_string());
|
||||
}
|
||||
|
||||
let membership = crate::observed_unread::load_membership(
|
||||
&app,
|
||||
&crate::observed_unread::ObservedUnreadScope {
|
||||
pubkey: owner,
|
||||
relay_url,
|
||||
},
|
||||
)?;
|
||||
let mut channels = classify_batch(&request, fetched, &membership);
|
||||
channels.extend(failures);
|
||||
Ok(UnreadCatchUpResponse { channels })
|
||||
}
|
||||
|
||||
// Shared production boundary: original per-channel filters, finite requests,
|
||||
// and error partitioning. Scope/membership checks remain in the command above.
|
||||
async fn fetch_channels(
|
||||
session: std::sync::Arc<crate::native_relay_client::RelaySession>,
|
||||
request: &UnreadCatchUpRequest,
|
||||
) -> Result<(Vec<FetchedChannel>, Vec<ChannelResult>), String> {
|
||||
let concurrency = std::sync::Arc::new(Semaphore::new(8));
|
||||
let mut pending = JoinSet::new();
|
||||
// One command replaces N renderer invokes while the shared session still
|
||||
@@ -163,7 +190,7 @@ pub(crate) async fn unread_catch_up(
|
||||
.acquire_owned()
|
||||
.await
|
||||
.map_err(|error| error.to_string())?;
|
||||
let session = session.handle();
|
||||
let session = std::sync::Arc::clone(&session);
|
||||
pending.spawn(async move {
|
||||
let _permit = permit;
|
||||
let kinds: &[u32] = if channel.channel_type == "dm" {
|
||||
@@ -217,23 +244,7 @@ pub(crate) async fn unread_catch_up(
|
||||
|
||||
fetched.sort_by_key(|item| item.order);
|
||||
|
||||
let current_keys = state.signing_keys()?;
|
||||
if current_keys.public_key().to_hex() != owner
|
||||
|| crate::relay::relay_ws_url_with_override(&state) != relay_url
|
||||
{
|
||||
return Err("unread catch-up scope changed while fetching".to_string());
|
||||
}
|
||||
|
||||
let membership = crate::observed_unread::load_membership(
|
||||
&app,
|
||||
&crate::observed_unread::ObservedUnreadScope {
|
||||
pubkey: owner,
|
||||
relay_url,
|
||||
},
|
||||
)?;
|
||||
let mut channels = classify_batch(&request, fetched, &membership);
|
||||
channels.extend(failures);
|
||||
Ok(UnreadCatchUpResponse { channels })
|
||||
Ok((fetched, failures))
|
||||
}
|
||||
|
||||
fn classify_batch(
|
||||
@@ -667,3 +678,7 @@ mod tests {
|
||||
assert_eq!(actual, expected);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "unread_catch_up_recovery_tests.rs"]
|
||||
mod recovery_tests;
|
||||
|
||||
201
desktop/src-tauri/src/unread_catch_up_recovery_tests.rs
Normal file
201
desktop/src-tauri/src/unread_catch_up_recovery_tests.rs
Normal file
@@ -0,0 +1,201 @@
|
||||
//! Socket refusal -> production unread fetch/filter -> batch classifier.
|
||||
use super::*;
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
|
||||
use tokio_tungstenite::tungstenite::Message;
|
||||
|
||||
#[tokio::test]
|
||||
async fn refused_unread_history_recovers_unique_events_with_original_filters() {
|
||||
let keys = Keys::generate();
|
||||
let owner = keys.public_key().to_hex();
|
||||
let author = Keys::generate();
|
||||
let root = "a".repeat(64);
|
||||
let make = |content: &str, signer: &Keys, at, tags: Vec<Tag>| {
|
||||
EventBuilder::new(Kind::Custom(9), content)
|
||||
.custom_created_at(Timestamp::from(at))
|
||||
.tags(tags)
|
||||
.sign_with_keys(signer)
|
||||
.unwrap()
|
||||
};
|
||||
let participation = make(
|
||||
"participation",
|
||||
&keys,
|
||||
42,
|
||||
vec![
|
||||
Tag::parse(["h", "stream"]).unwrap(),
|
||||
Tag::parse(["e", &root, "", "reply"]).unwrap(),
|
||||
],
|
||||
);
|
||||
let external = make(
|
||||
"reply",
|
||||
&author,
|
||||
43,
|
||||
vec![
|
||||
Tag::parse(["h", "stream"]).unwrap(),
|
||||
Tag::parse(["e", &root, "", "reply"]).unwrap(),
|
||||
],
|
||||
);
|
||||
let dm = make("dm", &author, 101, vec![Tag::parse(["h", "dm"]).unwrap()]);
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let url = format!("ws://{}", listener.local_addr().unwrap());
|
||||
let expected = [participation.clone(), external.clone(), dm.clone()];
|
||||
let server = tokio::spawn(async move {
|
||||
let (tcp, _) = listener.accept().await.unwrap();
|
||||
let mut ws = tokio_tungstenite::accept_async(tcp).await.unwrap();
|
||||
ws.send(Message::Text(
|
||||
serde_json::json!(["AUTH", "unread-test"])
|
||||
.to_string()
|
||||
.into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
let mut originals = std::collections::HashMap::new();
|
||||
let mut attempts = std::collections::HashMap::<String, usize>::new();
|
||||
let mut completed = 0;
|
||||
while let Some(Ok(Message::Text(text))) = ws.next().await {
|
||||
let frame: serde_json::Value = serde_json::from_str(&text).unwrap();
|
||||
let id = frame[1].as_str().unwrap_or_default();
|
||||
match frame[0].as_str() {
|
||||
Some("AUTH") => {
|
||||
ws.send(Message::Text(
|
||||
serde_json::json!(["OK", frame[1]["id"], true, ""])
|
||||
.to_string()
|
||||
.into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
Some("REQ") => {
|
||||
let channel = frame[2]["#h"][0].as_str().unwrap();
|
||||
let count = attempts.entry(channel.to_owned()).or_default();
|
||||
*count += 1;
|
||||
if channel == "denied" {
|
||||
ws.send(Message::Text(
|
||||
serde_json::json!(["CLOSED", id, "restricted: denied"])
|
||||
.to_string()
|
||||
.into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
continue;
|
||||
}
|
||||
assert_eq!(frame[2]["limit"], 1000);
|
||||
assert_eq!(frame[2]["since"], if channel == "dm" { 101 } else { 42 });
|
||||
assert_eq!(
|
||||
frame[2]["kinds"],
|
||||
if channel == "dm" {
|
||||
serde_json::json!([9, 40002, 45001, 45003, KIND_HUDDLE_STARTED])
|
||||
} else {
|
||||
serde_json::json!([9, 40002, 45001, 45003])
|
||||
}
|
||||
);
|
||||
if *count == 1 {
|
||||
originals.insert(
|
||||
channel.to_owned(),
|
||||
(frame.clone(), std::time::Instant::now()),
|
||||
);
|
||||
if channel == "stream" {
|
||||
ws.send(Message::Text(
|
||||
serde_json::json!(["EVENT", id, participation])
|
||||
.to_string()
|
||||
.into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
ws.send(Message::Text(
|
||||
serde_json::json!([
|
||||
"CLOSED",
|
||||
id,
|
||||
"rate-limited: quota exceeded; retry in 1s"
|
||||
])
|
||||
.to_string()
|
||||
.into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
} else {
|
||||
assert_eq!(*count, 2);
|
||||
let (original, refused) = &originals[channel];
|
||||
assert_eq!(&frame, original, "retry must keep exact filter and id");
|
||||
assert!(refused.elapsed() >= Duration::from_secs(1));
|
||||
let events = if channel == "dm" {
|
||||
vec![&dm]
|
||||
} else {
|
||||
vec![&participation, &external, &external]
|
||||
};
|
||||
for event in events {
|
||||
ws.send(Message::Text(
|
||||
serde_json::json!(["EVENT", id, event]).to_string().into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
ws.send(Message::Text(
|
||||
serde_json::json!(["EOSE", id]).to_string().into(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
completed += 1;
|
||||
}
|
||||
}
|
||||
Some("CLOSE") if completed == 2 => break,
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
assert_eq!(completed, 2);
|
||||
assert_eq!(attempts["denied"], 1);
|
||||
});
|
||||
let (session, _archive) = crate::native_relay_client::start(url, keys, None).await;
|
||||
let request = UnreadCatchUpRequest {
|
||||
self_pubkey: owner,
|
||||
muted_channel_ids: HashSet::new(),
|
||||
channels: vec![
|
||||
("stream", "stream", 41),
|
||||
("dm", "dm", 100),
|
||||
("denied", "stream", 41),
|
||||
]
|
||||
.into_iter()
|
||||
.map(|(id, channel_type, read_at)| CatchUpChannel {
|
||||
id: id.into(),
|
||||
channel_type: channel_type.into(),
|
||||
name: id.into(),
|
||||
read_at: Some(read_at),
|
||||
})
|
||||
.collect(),
|
||||
};
|
||||
let (fetched, failures) = fetch_channels(std::sync::Arc::clone(&session), &request)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(failures.len(), 1);
|
||||
assert!(
|
||||
matches!(&failures[0], ChannelResult::Error {channel_id, error} if channel_id == "denied" && error.contains("restricted:"))
|
||||
);
|
||||
assert_eq!(fetched.iter().map(|f| f.events.len()).sum::<usize>(), 3);
|
||||
let result = classify_batch(&request, fetched, &std::collections::HashMap::new());
|
||||
let ChannelResult::Success {
|
||||
observed_events,
|
||||
max_trigger,
|
||||
discovered,
|
||||
..
|
||||
} = &result[0]
|
||||
else {
|
||||
panic!("stream success")
|
||||
};
|
||||
assert_eq!(observed_events.len(), 1);
|
||||
assert_eq!(observed_events[0].id, expected[1].id.to_hex());
|
||||
assert!(observed_events[0].high_priority);
|
||||
assert_eq!(*max_trigger, 43);
|
||||
assert_eq!(discovered.participated, [root]);
|
||||
let ChannelResult::Success {
|
||||
observed_events, ..
|
||||
} = &result[1]
|
||||
else {
|
||||
panic!("dm success")
|
||||
};
|
||||
assert_eq!(observed_events.len(), 1);
|
||||
assert_eq!(observed_events[0].id, expected[2].id.to_hex());
|
||||
assert!(observed_events[0].counts_toward_app_badge);
|
||||
server.await.unwrap();
|
||||
session.shutdown();
|
||||
}
|
||||
Reference in New Issue
Block a user