+1

Cargo.lock

Reviewed

··· 3771 3771 version = "0.1.0" 3772 3772 dependencies = [ 3773 3773 "anyhow", 3774 3774 + "async-trait", 3774 3775 "bincode 2.0.1", 3775 3776 "cardinality-estimator", 3776 3777 "clap",

+1

ufos/Cargo.toml

Reviewed

··· 5 5 6 6 [dependencies] 7 7 anyhow = "1.0.97" 8 8 + async-trait = "0.1.88" 8 9 bincode = { version = "2.0.1", features = ["serde"] } 9 10 cardinality-estimator = { version = "1.0.2", features = ["with_serde"] } 10 11 clap = { version = "4.5.31", features = ["derive"] }

+2

ufos/src/error.rs

Reviewed

··· 36 36 EncodingError(#[from] EncodingError), 37 37 #[error("If you ever see this, there's a bug in the code. The error was stolen")] 38 38 Stolen, 39 39 + #[error("Failed to join tokio task: {0}")] 40 40 + JoinError(#[from] tokio::task::JoinError), 39 41 }

+1 -1

ufos/src/server.rs

Reviewed

··· 48 48 let failed_to_get = 49 49 |what| move |e| HttpError::for_internal_error(format!("failed to get {what}: {e:?}")); 50 50 51 51 - let storage_info = block_in_place(|| storage.get_storage_stats()) 51 51 + let storage_info = storage.get_storage_stats_a().await 52 52 .map_err(failed_to_get("storage info"))?; 53 53 54 54 let consumer = block_in_place(|| storage.get_consumer_info())

+3

ufos/src/storage.rs

Reviewed

··· 2 2 use crate::{error::StorageError, ConsumerInfo, Cursor, EventBatch, TopCollections, UFOsRecord}; 3 3 use jetstream::exports::{Did, Nsid}; 4 4 use std::path::Path; 5 5 + use async_trait::async_trait; 5 6 6 7 pub type StorageResult<T> = Result<T, StorageError>; 7 8 ··· 27 28 fn delete_account(&mut self, did: &Did) -> StorageResult<usize>; 28 29 } 29 30 31 31 + #[async_trait] 30 32 pub trait StoreReader: Send + Sync { 31 33 fn get_storage_stats(&self) -> StorageResult<serde_json::Value>; 34 34 + async fn get_storage_stats_a(&self) -> StorageResult<serde_json::Value>; 32 35 33 36 fn get_consumer_info(&self) -> StorageResult<ConsumerInfo>; 34 37

+6

ufos/src/storage_fjall.rs

Reviewed

··· 26 26 use std::time::{Duration, Instant, SystemTime}; 27 27 use tokio::sync::mpsc::Receiver; 28 28 use tokio::time::{interval_at, sleep}; 29 29 + use async_trait::async_trait; 29 30 30 31 /// Commit the RW batch immediately if this number of events have been read off the mod queue 31 32 const MAX_BATCHED_RW_EVENTS: usize = 18; ··· 300 301 } 301 302 } 302 303 304 304 + #[async_trait] 303 305 impl StoreReader for FjallReader { 304 306 fn get_storage_stats(&self) -> StorageResult<serde_json::Value> { 305 307 let rollup_cursor = ··· 312 314 "keyspace_sequence": self.keyspace.instant(), 313 315 "rollup_cursor": rollup_cursor, 314 316 })) 317 317 + } 318 318 + async fn get_storage_stats_a(&self) -> StorageResult<serde_json::Value> { 319 319 + let s = self.clone(); 320 320 + tokio::task::spawn_blocking(move || FjallReader::get_storage_stats(&s)).await? 315 321 } 316 322 317 323 fn get_consumer_info(&self) -> StorageResult<ConsumerInfo> {