diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 811253e4ac..9f4800743c 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -5,6 +5,7 @@ mod config; mod engram_fetch; mod filter; mod observer; +mod pending_store; mod pool; mod pool_lifecycle; mod queue; @@ -1553,8 +1554,18 @@ async fn tokio_main() -> Result<()> { let runtime_start_nonce = std::env::var("BUZZ_MANAGED_AGENT_START_NONCE").unwrap_or_default(); let dedup_mode = config.dedup_mode; - let mut queue = - EventQueue::new(dedup_mode).with_in_flight_deadline(config.max_turn_duration_secs); + let pending_store = pending_store::PendingStore::open(&pubkey_hex) + .map_err(|error| anyhow::anyhow!("open durable pending-work journal: {error}"))?; + let mut queue = EventQueue::new(dedup_mode) + .with_in_flight_deadline(config.max_turn_duration_secs) + .with_pending_store(pending_store); + let restored_event_ids = queue.pending_event_ids(); + if !restored_event_ids.is_empty() { + let rest = relay.rest_client(); + for event_id in &restored_event_ids { + pool::reaction_add(&rest, event_id, "👀").await; + } + } // Online means the harness can receive work, not merely that its socket is // connected. Publishing after channel subscriptions gives desktop callers @@ -2731,6 +2742,15 @@ async fn tokio_main() -> Result<()> { // explicitly shut them down here to reap child processes. If the grace // period expires, remaining tasks are aborted and fall back to // AcpClient::Drop (start_kill + try_wait — best-effort, not guaranteed). + let shutdown_batches: HashMap = pool + .task_map() + .values() + .filter_map(|meta| { + meta.recoverable_batch + .clone() + .map(|batch| (meta.agent_index, batch)) + }) + .collect(); let (rx_ref, js_ref) = pool.rx_and_join_set(); let shutdown_result = tokio::time::timeout(grace, async { loop { @@ -2745,6 +2765,11 @@ async fn tokio_main() -> Result<()> { maybe_result = rx_ref.recv() => { if let Some(mut pr) = maybe_result { let idx = pr.agent.index; + if matches!(pr.outcome, PromptOutcome::Ok(_)) { + if let Some(batch) = shutdown_batches.get(&idx) { + queue.finish_batch(batch); + } + } pr.agent.acp.shutdown().await; tracing::debug!(agent = idx, "reaped checked-out agent on shutdown"); } @@ -2762,6 +2787,11 @@ async fn tokio_main() -> Result<()> { // before tasks were aborted. while let Ok(mut pr) = pool.result_rx_try_recv() { let idx = pr.agent.index; + if matches!(pr.outcome, PromptOutcome::Ok(_)) { + if let Some(batch) = shutdown_batches.get(&idx) { + queue.finish_batch(batch); + } + } pr.agent.acp.shutdown().await; tracing::debug!(agent = idx, "reaped late-arriving agent on shutdown"); } @@ -3166,6 +3196,12 @@ fn handle_prompt_result( observer: Option, rest_client: Option<&relay::RestClient>, ) -> LoopAction { + let had_retry_batch = result.batch.is_some(); + let recoverable_batch = pool + .task_map() + .values() + .find(|meta| meta.agent_index == result.agent.index) + .and_then(|meta| meta.recoverable_batch.clone()); let before = pool.task_map().len(); let agent_index = result.agent.index; pool.task_map_mut() @@ -3227,6 +3263,7 @@ fn handle_prompt_result( config.max_turn_duration_secs ); spawn_failure_notice(rest_client, &batch, content); + queue.finish_batch(&batch); hard_timeout_fate_suffix = Some(" — dead-lettered (no recent activity)"); } else if matches!( result.outcome, @@ -3245,6 +3282,7 @@ fn handle_prompt_result( config.max_turn_duration_secs ); spawn_failure_notice(rest_client, &dead, content); + queue.finish_batch(&dead); hard_timeout_fate_suffix = Some(" — dead-lettered (retry budget exhausted)"); } else { hard_timeout_fate_suffix = Some(" — requeued for retry (recently active)"); @@ -3264,6 +3302,7 @@ fn handle_prompt_result( and then re-send." .to_string(); spawn_failure_notice(rest_client, &batch, content); + queue.finish_batch(&batch); } else if let Some(dead) = queue.requeue(batch) { let reason = match &result.outcome { PromptOutcome::Timeout(TimeoutKind::Idle) => "the turn timed out".to_string(), @@ -3278,6 +3317,7 @@ fn handle_prompt_result( "⚠️ I couldn't process the last request after multiple retries ({reason}). Please re-send if it's still needed." ); spawn_failure_notice(rest_client, &dead, content); + queue.finish_batch(&dead); } } else { tracing::debug!( @@ -3286,6 +3326,18 @@ fn handle_prompt_result( "dropping failed batch for removed channel" ); hard_timeout_fate_suffix = Some(" — batch dropped (channel removed)"); + queue.finish_batch(&batch); + } + } + + if matches!(result.outcome, PromptOutcome::Ok(_)) + || (matches!( + result.outcome, + PromptOutcome::Cancelled | PromptOutcome::CancelDrainTimeout(_) + ) && !had_retry_batch) + { + if let Some(batch) = recoverable_batch.as_ref() { + queue.finish_batch(batch); } } @@ -3547,7 +3599,9 @@ fn recover_panicked_agent( if !removed_channels.contains(&ch) { // Dead-letter on exhaustion is logged inside requeue(); a // panic path has no outcome to report, so no notice here. - let _ = queue.requeue(batch); + if let Some(dead) = queue.requeue(batch) { + queue.finish_batch(&dead); + } tracing::warn!("requeued batch for panicked agent {i}"); } else { tracing::debug!( diff --git a/crates/buzz-acp/src/pending_store.rs b/crates/buzz-acp/src/pending_store.rs new file mode 100644 index 0000000000..e695de4dd4 --- /dev/null +++ b/crates/buzz-acp/src/pending_store.rs @@ -0,0 +1,195 @@ +//! Durable journal for channel events accepted by the ACP queue. +//! +//! The relay subscription watermark intentionally starts at process startup, so +//! an event accepted immediately before a service restart is not replayed by +//! the relay. This journal closes that gap: an event is written atomically +//! before the harness publishes its "seen" reaction and is removed only after +//! the turn succeeds or the user receives a terminal failure. + +use nostr::Event; +use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; +use std::io; +use std::path::{Path, PathBuf}; +use uuid::Uuid; + +const STORE_VERSION: u32 = 1; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct StoredPendingEvent { + pub channel_id: Uuid, + pub event: Event, + pub prompt_tag: String, + pub accepted_at_nanos: u64, +} + +#[derive(Debug, Serialize, Deserialize)] +struct StoreFile { + version: u32, + events: BTreeMap, +} + +pub(crate) struct PendingStore { + path: PathBuf, + events: BTreeMap, +} + +impl PendingStore { + pub(crate) fn open(agent_pubkey: &str) -> io::Result { + let path = pending_store_path(agent_pubkey)?; + Self::open_path(path) + } + + pub(crate) fn open_path(path: PathBuf) -> io::Result { + let events = match std::fs::read(&path) { + Ok(bytes) => { + let stored: StoreFile = serde_json::from_slice(&bytes).map_err(|error| { + io::Error::new( + io::ErrorKind::InvalidData, + format!("invalid pending-work journal {}: {error}", path.display()), + ) + })?; + if stored.version != STORE_VERSION { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "unsupported pending-work journal version {} in {}", + stored.version, + path.display() + ), + )); + } + stored.events + } + Err(error) if error.kind() == io::ErrorKind::NotFound => BTreeMap::new(), + Err(error) => return Err(error), + }; + Ok(Self { path, events }) + } + + pub(crate) fn restored(&self) -> Vec { + let mut events = self.events.values().cloned().collect::>(); + events.sort_by_key(|event| event.accepted_at_nanos); + events + } + + pub(crate) fn record(&mut self, event: StoredPendingEvent) -> io::Result<()> { + let id = event.event.id.to_hex(); + let previous = self.events.insert(id.clone(), event); + if let Err(error) = self.persist() { + match previous { + Some(previous) => { + self.events.insert(id, previous); + } + None => { + self.events.remove(&id); + } + } + return Err(error); + } + Ok(()) + } + + pub(crate) fn remove<'a>(&mut self, ids: impl IntoIterator) -> io::Result<()> { + let removed: Vec<(String, StoredPendingEvent)> = ids + .into_iter() + .filter_map(|id| self.events.remove_entry(id)) + .collect(); + if removed.is_empty() { + return Ok(()); + } + if let Err(error) = self.persist() { + self.events.extend(removed); + return Err(error); + } + Ok(()) + } + + fn persist(&self) -> io::Result<()> { + let parent = self.path.parent().ok_or_else(|| { + io::Error::new(io::ErrorKind::InvalidInput, "pending journal has no parent") + })?; + std::fs::create_dir_all(parent)?; + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(parent, std::fs::Permissions::from_mode(0o700))?; + } + let bytes = serde_json::to_vec(&StoreFile { + version: STORE_VERSION, + events: self.events.clone(), + }) + .map_err(io::Error::other)?; + let temp = self.path.with_extension("json.tmp"); + let mut options = std::fs::OpenOptions::new(); + options.create(true).truncate(true).write(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + let mut file = options.open(&temp)?; + std::io::Write::write_all(&mut file, &bytes)?; + file.sync_all()?; + std::fs::rename(temp, &self.path) + } +} + +fn pending_store_path(agent_pubkey: &str) -> io::Result { + if let Some(path) = std::env::var_os("BUZZ_ACP_PENDING_STORE") { + return Ok(PathBuf::from(path)); + } + let home = std::env::var_os("HOME").ok_or_else(|| { + io::Error::new( + io::ErrorKind::NotFound, + "HOME is unset and BUZZ_ACP_PENDING_STORE was not provided", + ) + })?; + Ok(Path::new(&home) + .join(".local/state/buzz-acp") + .join(format!("pending-{agent_pubkey}.json"))) +} + +#[cfg(test)] +mod tests { + use super::*; + use nostr::{EventBuilder, Keys}; + + fn event(content: &str) -> Event { + EventBuilder::text_note(content) + .sign_with_keys(&Keys::generate()) + .expect("sign event") + } + + #[test] + fn journal_round_trips_and_removes_atomically() { + let dir = std::env::temp_dir().join(format!("buzz-acp-pending-{}", Uuid::new_v4())); + std::fs::create_dir_all(&dir).expect("create tempdir"); + let path = dir.join("pending.json"); + let mut store = PendingStore::open_path(path.clone()).expect("open"); + let event = event("survive restart"); + let id = event.id.to_hex(); + store + .record(StoredPendingEvent { + channel_id: Uuid::new_v4(), + event, + prompt_tag: "@mention".into(), + accepted_at_nanos: 1, + }) + .expect("record"); + drop(store); + + let mut restored = PendingStore::open_path(path.clone()).expect("reopen"); + assert_eq!(restored.restored().len(), 1); + restored.remove([id.as_str()]).expect("remove"); + drop(restored); + assert_eq!( + PendingStore::open_path(path) + .expect("final reopen") + .restored() + .len(), + 0 + ); + std::fs::remove_dir_all(dir).expect("remove tempdir"); + } +} diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 5c960de202..502f778867 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -19,6 +19,7 @@ use std::time::{Duration, Instant}; use uuid::Uuid; use crate::config::DedupMode; +use crate::pending_store::{PendingStore, StoredPendingEvent}; /// Maximum events queued per channel before oldest events are dropped. const MAX_PENDING_PER_CHANNEL: usize = 500; @@ -168,6 +169,7 @@ pub struct EventQueue { /// Must be strictly greater than `max_turn_duration` so a turn running to /// the hard cap returns via `mark_complete` before the backstop fires. in_flight_deadline: Duration, + pending_store: Option, } impl EventQueue { @@ -189,9 +191,34 @@ impl EventQueue { cancel_reasons: HashMap::new(), withheld_native_steer: HashMap::new(), in_flight_deadline: Duration::from_secs(DEFAULT_IN_FLIGHT_DEADLINE_SECS), + pending_store: None, } } + /// Attach durable pending-work storage and restore unfinished events. + pub(crate) fn with_pending_store(mut self, store: PendingStore) -> Self { + for stored in store.restored() { + self.queues + .entry(stored.channel_id) + .or_default() + .push_back(QueuedEvent { + channel_id: stored.channel_id, + event: stored.event, + received_at: Instant::now(), + prompt_tag: stored.prompt_tag, + }); + } + if !self.queues.is_empty() { + tracing::warn!( + events = self.queues.values().map(VecDeque::len).sum::(), + channels = self.queues.len(), + "restored unfinished work from durable pending journal" + ); + } + self.pending_store = Some(store); + self + } + /// Set the in-flight backstop deadline from the configured max turn /// duration, preserving the 100s buffer for cancel-drain grace + respawn. pub fn with_in_flight_deadline(mut self, max_turn_duration_secs: u64) -> Self { @@ -237,20 +264,66 @@ impl EventQueue { ); return false; } + if let Some(store) = self.pending_store.as_mut() { + if let Err(error) = store.record(StoredPendingEvent { + channel_id: event.channel_id, + event: event.event.clone(), + prompt_tag: event.prompt_tag.clone(), + accepted_at_nanos: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + .min(u64::MAX as u128) as u64, + }) { + tracing::error!( + channel_id = %event.channel_id, + event_id = %event.event.id, + %error, + "refusing to acknowledge event because durable queue write failed" + ); + return false; + } + } let queue = self.queues.entry(event.channel_id).or_default(); // Enforce per-channel depth cap: drop oldest to make room. - if queue.len() >= MAX_PENDING_PER_CHANNEL { - queue.pop_front(); + let dropped_id = if queue.len() >= MAX_PENDING_PER_CHANNEL { + let dropped = queue.pop_front().map(|event| event.event.id.to_hex()); tracing::warn!( channel_id = %event.channel_id, limit = MAX_PENDING_PER_CHANNEL, "queue depth cap reached — dropped oldest event" ); - } + dropped + } else { + None + }; queue.push_back(event); + if let Some(dropped_id) = dropped_id { + self.remove_pending_ids([dropped_id]); + } true } + /// Permanently finish every event represented by a completed/dead batch. + pub fn finish_batch(&mut self, batch: &FlushBatch) { + let ids = batch + .events + .iter() + .chain(batch.cancelled_events.iter()) + .map(|event| event.event.id.to_hex()) + .collect::>(); + self.remove_pending_ids(ids); + } + + fn remove_pending_ids(&mut self, ids: impl IntoIterator) { + let ids = ids.into_iter().collect::>(); + if let Some(store) = self.pending_store.as_mut() { + if let Err(error) = store.remove(ids.iter().map(String::as_str)) { + tracing::error!(%error, "failed to remove completed events from durable pending journal"); + } + } + } + /// Try to flush the next batch. /// /// Returns `None` if all non-in-flight, non-throttled queues are empty. @@ -472,27 +545,30 @@ impl EventQueue { "requeueing failed batch with backoff" ); - let queue = self.queues.entry(channel_id).or_default(); - // Push to front in reverse order so original order is preserved. - for be in batch.events.into_iter().rev() { - queue.push_front(QueuedEvent { - channel_id, - event: be.event, - prompt_tag: be.prompt_tag, - received_at: be.received_at, // preserve original timestamp (#46) - }); - } - // Enforce per-channel cap: trim oldest (back) events if requeue pushed - // the queue over the limit. Without this, repeated requeue+push cycles - // can grow the queue unboundedly. - while queue.len() > MAX_PENDING_PER_CHANNEL { - queue.pop_back(); - tracing::warn!( - channel_id = %channel_id, - limit = MAX_PENDING_PER_CHANNEL, - "requeue overflow — dropped oldest event to enforce cap" - ); + let mut dropped_ids = Vec::new(); + { + let queue = self.queues.entry(channel_id).or_default(); + // Push to front in reverse order so original order is preserved. + for be in batch.events.into_iter().rev() { + queue.push_front(QueuedEvent { + channel_id, + event: be.event, + prompt_tag: be.prompt_tag, + received_at: be.received_at, // preserve original timestamp (#46) + }); + } + while queue.len() > MAX_PENDING_PER_CHANNEL { + if let Some(dropped) = queue.pop_back() { + dropped_ids.push(dropped.event.id.to_hex()); + } + tracing::warn!( + channel_id = %channel_id, + limit = MAX_PENDING_PER_CHANNEL, + "requeue overflow — dropped oldest event to enforce cap" + ); + } } + self.remove_pending_ids(dropped_ids); self.retry_after.insert(channel_id, Instant::now() + delay); None } @@ -507,25 +583,30 @@ impl EventQueue { /// caller must call `mark_complete` separately. pub fn requeue_preserve_timestamps(&mut self, batch: FlushBatch) { let channel_id = batch.channel_id; - let queue = self.queues.entry(channel_id).or_default(); - // Push to front in reverse order so original order is preserved. - for be in batch.events.into_iter().rev() { - queue.push_front(QueuedEvent { - channel_id, - event: be.event, - prompt_tag: be.prompt_tag, - received_at: be.received_at, - }); - } - // Enforce per-channel cap: trim newest (back) events if over limit. - while queue.len() > MAX_PENDING_PER_CHANNEL { - queue.pop_back(); - tracing::warn!( - channel_id = %channel_id, - limit = MAX_PENDING_PER_CHANNEL, - "requeue_preserve overflow — dropped newest event to enforce cap" - ); + let mut dropped_ids = Vec::new(); + { + let queue = self.queues.entry(channel_id).or_default(); + // Push to front in reverse order so original order is preserved. + for be in batch.events.into_iter().rev() { + queue.push_front(QueuedEvent { + channel_id, + event: be.event, + prompt_tag: be.prompt_tag, + received_at: be.received_at, + }); + } + while queue.len() > MAX_PENDING_PER_CHANNEL { + if let Some(dropped) = queue.pop_back() { + dropped_ids.push(dropped.event.id.to_hex()); + } + tracing::warn!( + channel_id = %channel_id, + limit = MAX_PENDING_PER_CHANNEL, + "requeue_preserve overflow — dropped newest event to enforce cap" + ); + } } + self.remove_pending_ids(dropped_ids); } /// Requeue a cancelled batch so its events appear as `cancelled_events` @@ -595,6 +676,15 @@ impl EventQueue { self.queues.len() } + /// Event IDs restored at startup, used to restore the visible seen marker. + pub(crate) fn pending_event_ids(&self) -> Vec { + self.queues + .values() + .flatten() + .map(|event| event.event.id.to_hex()) + .collect() + } + /// Number of queued events for a specific channel. Test-only. #[cfg(test)] pub fn queued_event_count(&self, channel_id: &Uuid) -> usize { @@ -623,16 +713,21 @@ impl EventQueue { /// Returns the event IDs of dropped events so the caller can clean up /// any reactions (👀) that were added at queue-push time. pub fn drain_channel(&mut self, channel_id: Uuid) -> Vec { - let ids = self + let mut ids: Vec = self .queues .remove(&channel_id) .map(|q| q.into_iter().map(|e| e.event.id.to_hex()).collect()) .unwrap_or_default(); self.retry_after.remove(&channel_id); self.retry_counts.remove(&channel_id); - self.cancelled_batches.remove(&channel_id); + if let Some(cancelled) = self.cancelled_batches.remove(&channel_id) { + ids.extend(cancelled.into_iter().map(|event| event.event.id.to_hex())); + } self.cancel_reasons.remove(&channel_id); - self.withheld_native_steer.remove(&channel_id); + if let Some(withheld) = self.withheld_native_steer.remove(&channel_id) { + ids.extend(withheld.into_iter().map(|event| event.event.id.to_hex())); + } + self.remove_pending_ids(ids.iter().cloned()); // Preserve in_flight_channels AND in_flight_deadlines: the in-flight // task will eventually complete (calling mark_complete) or the deadline // will expire (auto-cleaning the channel). Removing deadlines without @@ -722,16 +817,22 @@ impl EventQueue { // Push to FRONT so original `received_at` keeps the event at the head // of the channel's queue. Per-channel cap is enforced below in case // a flood of events arrived during the ack window. - let queue = self.queues.entry(channel_id).or_default(); - queue.push_front(qe); - while queue.len() > MAX_PENDING_PER_CHANNEL { - queue.pop_back(); - tracing::warn!( - channel_id = %channel_id, - limit = MAX_PENDING_PER_CHANNEL, - "release_native_steer overflow — dropped newest event to enforce cap" - ); + let mut dropped_ids = Vec::new(); + { + let queue = self.queues.entry(channel_id).or_default(); + queue.push_front(qe); + while queue.len() > MAX_PENDING_PER_CHANNEL { + if let Some(dropped) = queue.pop_back() { + dropped_ids.push(dropped.event.id.to_hex()); + } + tracing::warn!( + channel_id = %channel_id, + limit = MAX_PENDING_PER_CHANNEL, + "release_native_steer overflow — dropped newest event to enforce cap" + ); + } } + self.remove_pending_ids(dropped_ids); } /// Drop a specific event by id from both the side table and the main @@ -747,6 +848,7 @@ impl EventQueue { self.withheld_native_steer.remove(&channel_id); } } + self.remove_pending_ids([event_id.to_string()]); if let Some(q) = self.queues.get_mut(&channel_id) { q.retain(|qe| qe.event.id.to_hex() != event_id); if q.is_empty() { @@ -773,18 +875,24 @@ impl EventQueue { return; }; let n = entries.len(); - let queue = self.queues.entry(channel_id).or_default(); - for qe in entries.into_iter().rev() { - queue.push_front(qe); - } - while queue.len() > MAX_PENDING_PER_CHANNEL { - queue.pop_back(); - tracing::warn!( - channel_id = %channel_id, - limit = MAX_PENDING_PER_CHANNEL, - "withheld-steer recovery overflow — dropped newest event to enforce cap" - ); + let mut dropped_ids = Vec::new(); + { + let queue = self.queues.entry(channel_id).or_default(); + for qe in entries.into_iter().rev() { + queue.push_front(qe); + } + while queue.len() > MAX_PENDING_PER_CHANNEL { + if let Some(dropped) = queue.pop_back() { + dropped_ids.push(dropped.event.id.to_hex()); + } + tracing::warn!( + channel_id = %channel_id, + limit = MAX_PENDING_PER_CHANNEL, + "withheld-steer recovery overflow — dropped newest event to enforce cap" + ); + } } + self.remove_pending_ids(dropped_ids); tracing::warn!( channel_id = %channel_id, recovered = n, @@ -4762,3 +4870,40 @@ mod tests { ); } } +#[test] +fn accepted_event_survives_queue_recreation_until_finished() { + let dir = std::env::temp_dir().join(format!("buzz-acp-queue-{}", Uuid::new_v4())); + std::fs::create_dir_all(&dir).expect("create tempdir"); + let path = dir.join("pending.json"); + let channel_id = Uuid::new_v4(); + let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "survive restart") + .sign_with_keys(&nostr::Keys::generate()) + .expect("sign event"); + let queued = QueuedEvent { + channel_id, + event, + received_at: Instant::now(), + prompt_tag: "test".into(), + }; + let event_id = queued.event.id.to_hex(); + + let store = PendingStore::open_path(path.clone()).expect("open store"); + let mut before = EventQueue::new(DedupMode::Queue).with_pending_store(store); + assert!(before.push(queued)); + let in_flight = before.flush_next().expect("dispatch before restart"); + assert_eq!(in_flight.events[0].event.id.to_hex(), event_id); + drop(before); + + let store = PendingStore::open_path(path.clone()).expect("reopen store"); + let mut after = EventQueue::new(DedupMode::Queue).with_pending_store(store); + let recovered = after.flush_next().expect("restore after restart"); + assert_eq!(recovered.events[0].event.id.to_hex(), event_id); + after.finish_batch(&recovered); + drop(after); + + let store = PendingStore::open_path(path).expect("final reopen"); + let mut final_queue = EventQueue::new(DedupMode::Queue).with_pending_store(store); + assert!(final_queue.flush_next().is_none()); + drop(final_queue); + std::fs::remove_dir_all(dir).expect("remove tempdir"); +}