diff --git a/CHANGELOG.md b/CHANGELOG.md index 8b45bb6..9ad4f78 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - **Gmail** — `void gmail search` and `void gmail thread` read from the local INBOX store when a usable body is already synced. `--live` forces the Gmail API (`in:sent`, drafts, and unsynced mail still go to the network). +- **Gmail** — Cross-process token bucket (~90 requests / 60s per account, stored in SQLite) so the CLI and sync daemon share quota instead of stampeding after a 429. ### Fixed diff --git a/crates/void-core/src/db/database_access.rs b/crates/void-core/src/db/database_access.rs index a963875..6347807 100644 --- a/crates/void-core/src/db/database_access.rs +++ b/crates/void-core/src/db/database_access.rs @@ -1,9 +1,13 @@ //! `Database` methods: hook logs and delegated CRUD entry points. +use std::time::Duration; + use crate::error::DbError; use crate::models::{CalendarEvent, Contact, Conversation, Message}; -use super::{conversations, directory, events, hook_logs, messages, mute_sync, Database}; +use super::{ + conversations, directory, events, hook_logs, messages, mute_sync, rate_limit, Database, +}; impl Database { pub fn insert_hook_log(&self, log: &crate::hooks::HookLogInsert<'_>) -> Result<(), DbError> { @@ -562,6 +566,21 @@ impl Database { mute_sync::set_sync_state(&*self.conn()?, connection_id, key, value) } + /// Take one token from a named bucket. Returns how long to sleep (zero = go). + /// + /// Uses `BEGIN IMMEDIATE` so two processes sharing the store serialize. + /// Does not sleep; the caller waits outside the write lock. + pub fn take_rate_token( + &self, + connection_id: &str, + key: &str, + capacity: f64, + refill_per_sec: f64, + ) -> Result { + let mut conn = self.conn()?; + rate_limit::take_token(&mut conn, connection_id, key, capacity, refill_per_sec) + } + pub fn rename_connection(&self, old_id: &str, new_id: &str) -> Result<(), DbError> { mute_sync::rename_connection(&*self.conn()?, old_id, new_id) } diff --git a/crates/void-core/src/db/mod.rs b/crates/void-core/src/db/mod.rs index fe24b56..33482ae 100644 --- a/crates/void-core/src/db/mod.rs +++ b/crates/void-core/src/db/mod.rs @@ -10,6 +10,7 @@ mod events; mod hook_logs; mod messages; mod mute_sync; +mod rate_limit; mod row; mod schema; mod search; diff --git a/crates/void-core/src/db/rate_limit.rs b/crates/void-core/src/db/rate_limit.rs new file mode 100644 index 0000000..6239fe5 --- /dev/null +++ b/crates/void-core/src/db/rate_limit.rs @@ -0,0 +1,78 @@ +//! Cross-process token bucket stored in `sync_state`. +//! +//! Two processes opening the same store (CLI + sync daemon) serialize on +//! `BEGIN IMMEDIATE` and share one bucket. The function never sleeps: it +//! returns how long the caller should wait so the write lock is not held +//! across a sleep. + +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use rusqlite::{params, Connection, OptionalExtension, TransactionBehavior}; +use serde::{Deserialize, Serialize}; + +use crate::error::DbError; + +#[derive(Debug, Serialize, Deserialize)] +struct Bucket { + tokens: f64, + /// Unix seconds, fractional. + updated: f64, +} + +fn now_secs() -> f64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs_f64()) + .unwrap_or(0.0) +} + +pub(super) fn take_token( + conn: &mut Connection, + connection_id: &str, + key: &str, + capacity: f64, + refill_per_sec: f64, +) -> Result { + if capacity <= 0.0 || refill_per_sec <= 0.0 { + return Ok(Duration::ZERO); + } + + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let raw: Option = tx + .query_row( + "SELECT value FROM sync_state WHERE connection_id = ?1 AND key = ?2", + params![connection_id, key], + |row| row.get(0), + ) + .optional()?; + + let now = now_secs(); + let mut bucket = raw + .as_deref() + .and_then(|s| serde_json::from_str::(s).ok()) + .unwrap_or(Bucket { + tokens: capacity, + updated: now, + }); + + let elapsed = (now - bucket.updated).max(0.0); + bucket.tokens = (bucket.tokens + elapsed * refill_per_sec).min(capacity); + bucket.updated = now; + + let wait = if bucket.tokens >= 1.0 { + bucket.tokens -= 1.0; + Duration::ZERO + } else { + let needed = 1.0 - bucket.tokens; + Duration::from_secs_f64((needed / refill_per_sec).clamp(0.0, 3600.0)) + }; + + let value = serde_json::to_string(&bucket).expect("bucket serializes"); + tx.execute( + "INSERT INTO sync_state (connection_id, key, value) VALUES (?1, ?2, ?3) + ON CONFLICT(connection_id, key) DO UPDATE SET value = excluded.value", + params![connection_id, key, value], + )?; + tx.commit()?; + Ok(wait) +} diff --git a/crates/void-core/src/db/tests/mod.rs b/crates/void-core/src/db/tests/mod.rs index 7cd3f18..d9efc24 100644 --- a/crates/void-core/src/db/tests/mod.rs +++ b/crates/void-core/src/db/tests/mod.rs @@ -6,5 +6,6 @@ mod crud; mod dedup; mod fixtures; mod mute; +mod rate_limit; mod saved; mod search; diff --git a/crates/void-core/src/db/tests/rate_limit.rs b/crates/void-core/src/db/tests/rate_limit.rs new file mode 100644 index 0000000..a4a7972 --- /dev/null +++ b/crates/void-core/src/db/tests/rate_limit.rs @@ -0,0 +1,53 @@ +use std::time::Duration; + +use super::fixtures::*; + +#[test] +fn take_rate_token_waits_when_empty() { + let db = test_db(); + assert!(db.take_rate_token("g1", "k", 1.0, 0.01).unwrap().is_zero()); + let wait = db.take_rate_token("g1", "k", 1.0, 0.01).unwrap(); + assert!(wait > Duration::from_secs(1)); +} + +#[test] +fn take_rate_token_refills_from_stale_timestamp() { + let db = test_db(); + assert!(db.take_rate_token("g1", "k", 1.0, 1.0).unwrap().is_zero()); + let past = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs_f64() + - 10.0; + db.set_sync_state("g1", "k", &format!(r#"{{"tokens":0.0,"updated":{past}}}"#)) + .unwrap(); + assert!(db.take_rate_token("g1", "k", 1.0, 1.0).unwrap().is_zero()); +} + +#[test] +fn take_rate_token_is_per_connection() { + let db = test_db(); + assert!(db.take_rate_token("g1", "k", 1.0, 0.01).unwrap().is_zero()); + assert!(db.take_rate_token("g2", "k", 1.0, 0.01).unwrap().is_zero()); +} + +#[test] +fn take_rate_token_is_shared_across_db_handles() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("void.db"); + let db1 = crate::db::Database::open(&path).unwrap(); + let db2 = crate::db::Database::open(&path).unwrap(); + assert!(db1.take_rate_token("g", "k", 1.0, 0.01).unwrap().is_zero()); + let wait = db2.take_rate_token("g", "k", 1.0, 0.01).unwrap(); + assert!(wait > Duration::from_secs(1)); +} + +#[test] +fn take_rate_token_corrupt_json_fails_open_as_full_bucket() { + let db = test_db(); + db.set_sync_state("g1", "k", "not-json").unwrap(); + assert!(db.take_rate_token("g1", "k", 2.0, 1.0).unwrap().is_zero()); + assert!(db.take_rate_token("g1", "k", 2.0, 1.0).unwrap().is_zero()); + let wait = db.take_rate_token("g1", "k", 2.0, 1.0).unwrap(); + assert!(wait > Duration::ZERO); +} diff --git a/crates/void-gmail/src/api/client.rs b/crates/void-gmail/src/api/client.rs index 011911f..44fa9b3 100644 --- a/crates/void-gmail/src/api/client.rs +++ b/crates/void-gmail/src/api/client.rs @@ -1,8 +1,11 @@ +use std::path::Path; use std::time::Duration; use crate::error::GmailError; use tracing::{debug, info}; +use void_core::db::Database; +use super::rate_limit::StoreRateLimiter; use super::retry::{RetryPolicy, SendRetrying}; use super::types::{ @@ -29,6 +32,7 @@ pub struct GmailApiClient { access_token: String, base_url: String, retry: RetryPolicy, + limiter: Option, } impl GmailApiClient { @@ -38,6 +42,7 @@ impl GmailApiClient { access_token: access_token.to_string(), base_url: DEFAULT_BASE_URL.to_string(), retry: RetryPolicy::default(), + limiter: None, } } @@ -48,9 +53,37 @@ impl GmailApiClient { access_token: access_token.to_string(), base_url: base_url.to_string(), retry: RetryPolicy::fast(), + limiter: None, } } + /// Share a SQLite token bucket with other processes using the same store. + /// + /// Missing or unreadable `void.db` fails open (no limiter). Tests using + /// [`Self::with_base_url`] never attach one. + pub fn with_store_limiter(self, store_path: &Path, connection_id: &str) -> Self { + let path = store_path.join("void.db"); + if !path.exists() { + return self; + } + match Database::open(&path) { + Ok(db) => self.with_limiter(StoreRateLimiter::new(db, connection_id.to_string())), + Err(e) => { + tracing::warn!( + error = %e, + path = %path.display(), + "gmail: rate limiter unavailable, failing open" + ); + self + } + } + } + + fn with_limiter(mut self, limiter: StoreRateLimiter) -> Self { + self.limiter = Some(limiter); + self + } + pub fn set_token(&mut self, token: &str) { self.access_token = token.to_string(); } @@ -61,7 +94,7 @@ impl GmailApiClient { .http .get(format!("{}/gmail/v1/users/me/profile", self.base_url)) .bearer_auth(&self.access_token) - .send_retrying(&self.retry) + .send_retrying(&self.retry, self.limiter.as_ref()) .await? .json() .await?; @@ -99,7 +132,7 @@ impl GmailApiClient { .get(format!("{}/gmail/v1/users/me/messages", self.base_url)) .bearer_auth(&self.access_token) .query(¶ms) - .send_retrying(&self.retry) + .send_retrying(&self.retry, self.limiter.as_ref()) .await? .error_for_status()?; let resp: MessageListResponse = resp.json().await?; @@ -122,7 +155,7 @@ impl GmailApiClient { )) .bearer_auth(&self.access_token) .query(&[("format", "full")]) - .send_retrying(&self.retry) + .send_retrying(&self.retry, self.limiter.as_ref()) .await? .error_for_status()?; let resp: GmailMessage = resp.json().await?; @@ -153,7 +186,7 @@ impl GmailApiClient { .get(format!("{}/gmail/v1/users/me/history", self.base_url)) .bearer_auth(&self.access_token) .query(¶ms) - .send_retrying(&self.retry) + .send_retrying(&self.retry, self.limiter.as_ref()) .await?; // Gmail returns 404 once the startHistoryId is too old (history is // only kept for a limited window). Surface that distinctly so the @@ -215,7 +248,7 @@ impl GmailApiClient { )) .bearer_auth(&self.access_token) .json(&body) - .send_retrying(&self.retry) + .send_retrying(&self.retry, self.limiter.as_ref()) .await? .json() .await?; @@ -231,7 +264,7 @@ impl GmailApiClient { .post(format!("{}/gmail/v1/users/me/messages/send", self.base_url)) .bearer_auth(&self.access_token) .json(&body) - .send_retrying(&self.retry) + .send_retrying(&self.retry, self.limiter.as_ref()) .await? .json() .await?; @@ -249,7 +282,7 @@ impl GmailApiClient { )) .bearer_auth(&self.access_token) .query(&[("format", "full")]) - .send_retrying(&self.retry) + .send_retrying(&self.retry, self.limiter.as_ref()) .await? .error_for_status()? .json() @@ -272,7 +305,7 @@ impl GmailApiClient { self.base_url )) .bearer_auth(&self.access_token) - .send_retrying(&self.retry) + .send_retrying(&self.retry, self.limiter.as_ref()) .await? .error_for_status()? .json() @@ -287,7 +320,7 @@ impl GmailApiClient { .http .get(format!("{}/gmail/v1/users/me/labels", self.base_url)) .bearer_auth(&self.access_token) - .send_retrying(&self.retry) + .send_retrying(&self.retry, self.limiter.as_ref()) .await? .error_for_status()? .json() @@ -321,7 +354,7 @@ impl GmailApiClient { )) .bearer_auth(&self.access_token) .json(&body) - .send_retrying(&self.retry) + .send_retrying(&self.retry, self.limiter.as_ref()) .await? .error_for_status()? .json() @@ -354,7 +387,7 @@ impl GmailApiClient { )) .bearer_auth(&self.access_token) .json(&body) - .send_retrying(&self.retry) + .send_retrying(&self.retry, self.limiter.as_ref()) .await? .error_for_status()?; debug!("gmail: batch_modify ok"); @@ -368,7 +401,7 @@ impl GmailApiClient { .get(format!("{}/gmail/v1/users/me/drafts", self.base_url)) .bearer_auth(&self.access_token) .query(&[("maxResults", max_results.to_string())]) - .send_retrying(&self.retry) + .send_retrying(&self.retry, self.limiter.as_ref()) .await? .error_for_status()? .json() @@ -388,7 +421,7 @@ impl GmailApiClient { )) .bearer_auth(&self.access_token) .query(&[("format", "full")]) - .send_retrying(&self.retry) + .send_retrying(&self.retry, self.limiter.as_ref()) .await? .error_for_status()? .json() @@ -413,7 +446,7 @@ impl GmailApiClient { .post(format!("{}/gmail/v1/users/me/drafts", self.base_url)) .bearer_auth(&self.access_token) .json(&body) - .send_retrying(&self.retry) + .send_retrying(&self.retry, self.limiter.as_ref()) .await? .error_for_status()? .json() @@ -435,7 +468,7 @@ impl GmailApiClient { )) .bearer_auth(&self.access_token) .json(&body) - .send_retrying(&self.retry) + .send_retrying(&self.retry, self.limiter.as_ref()) .await? .error_for_status()? .json() @@ -452,7 +485,7 @@ impl GmailApiClient { self.base_url )) .bearer_auth(&self.access_token) - .send_retrying(&self.retry) + .send_retrying(&self.retry, self.limiter.as_ref()) .await? .error_for_status()?; debug!(draft_id, "gmail: delete_draft ok"); @@ -469,7 +502,7 @@ impl GmailApiClient { self.base_url )) .bearer_auth(&self.access_token) - .send_retrying(&self.retry) + .send_retrying(&self.retry, self.limiter.as_ref()) .await?; let resp: SendAsListResponse = Self::json_or_scope_error(resp).await?; let count = resp.send_as.as_ref().map(|s| s.len()).unwrap_or(0); @@ -488,7 +521,7 @@ impl GmailApiClient { self.base_url )) .bearer_auth(&self.access_token) - .send_retrying(&self.retry) + .send_retrying(&self.retry, self.limiter.as_ref()) .await?; let resp: SendAsAlias = Self::json_or_scope_error(resp).await?; debug!(send_as_email, "gmail: get_send_as ok"); diff --git a/crates/void-gmail/src/api/mod.rs b/crates/void-gmail/src/api/mod.rs index 2e9ad48..efbbc61 100644 --- a/crates/void-gmail/src/api/mod.rs +++ b/crates/void-gmail/src/api/mod.rs @@ -1,5 +1,6 @@ mod client; mod message; +mod rate_limit; mod retry; mod types; diff --git a/crates/void-gmail/src/api/rate_limit.rs b/crates/void-gmail/src/api/rate_limit.rs new file mode 100644 index 0000000..a6c3572 --- /dev/null +++ b/crates/void-gmail/src/api/rate_limit.rs @@ -0,0 +1,48 @@ +//! Process-shared Gmail quota bucket (CLI + sync daemon, same store). + +use std::time::Duration; + +use tracing::{debug, warn}; +use void_core::db::Database; + +const CAPACITY: f64 = 90.0; +const REFILL_PER_SEC: f64 = 90.0 / 60.0; +const KEY: &str = "gmail_rate_limit"; +const MAX_WAIT: Duration = Duration::from_secs(60); + +/// Token bucket persisted in SQLite `sync_state`. +pub struct StoreRateLimiter { + db: Database, + connection_id: String, +} + +impl StoreRateLimiter { + pub fn new(db: Database, connection_id: String) -> Self { + Self { db, connection_id } + } + + /// Wait until a token is available. Errors (and a missing store) fail open. + pub async fn acquire(&self) { + loop { + match self + .db + .take_rate_token(&self.connection_id, KEY, CAPACITY, REFILL_PER_SEC) + { + Ok(wait) if wait.is_zero() => return, + Ok(wait) => { + let sleep = wait.min(MAX_WAIT); + debug!( + wait_ms = sleep.as_millis() as u64, + connection_id = %self.connection_id, + "gmail: waiting for rate-limit token" + ); + tokio::time::sleep(sleep).await; + } + Err(e) => { + warn!(error = %e, "gmail: rate limiter failed open"); + return; + } + } + } + } +} diff --git a/crates/void-gmail/src/api/retry.rs b/crates/void-gmail/src/api/retry.rs index cd61d85..a34b662 100644 --- a/crates/void-gmail/src/api/retry.rs +++ b/crates/void-gmail/src/api/retry.rs @@ -179,9 +179,13 @@ fn quota_wait( pub async fn send_with_retry( req: reqwest::RequestBuilder, policy: &RetryPolicy, + limiter: Option<&super::rate_limit::StoreRateLimiter>, ) -> Result { let mut attempt = 1u32; loop { + if let Some(limiter) = limiter { + limiter.acquire().await; + } let clone = req.try_clone(); let is_last = attempt >= policy.max_attempts || clone.is_none(); @@ -242,12 +246,13 @@ pub async fn send_with_retry( } /// Lets a call site opt into retrying by replacing `.send()` with -/// `.send_retrying(&self.retry)`, keeping the rest of the chain untouched. +/// `.send_retrying(&self.retry, self.limiter.as_ref())`. pub trait SendRetrying { /// Send with the given retry policy. See [`send_with_retry`]. fn send_retrying( self, policy: &RetryPolicy, + limiter: Option<&super::rate_limit::StoreRateLimiter>, ) -> impl std::future::Future>; } @@ -255,8 +260,9 @@ impl SendRetrying for reqwest::RequestBuilder { async fn send_retrying( self, policy: &RetryPolicy, + limiter: Option<&super::rate_limit::StoreRateLimiter>, ) -> Result { - send_with_retry(self, policy).await + send_with_retry(self, policy, limiter).await } } diff --git a/crates/void-gmail/src/api/tests.rs b/crates/void-gmail/src/api/tests.rs index 1521bca..804c6e3 100644 --- a/crates/void-gmail/src/api/tests.rs +++ b/crates/void-gmail/src/api/tests.rs @@ -602,6 +602,7 @@ async fn get_thread_403_rate_limit_preserves_status_and_body_after_retries() { let resp = retry::send_with_retry( reqwest::Client::new().get(format!("{}/gmail/v1/users/me/threads/t403", server.uri())), &RetryPolicy::fast(), + None, ) .await .expect("retries exhausted, not a transport error"); diff --git a/crates/void-gmail/src/connector/api_methods.rs b/crates/void-gmail/src/connector/api_methods.rs index c9f6c30..97a189d 100644 --- a/crates/void-gmail/src/connector/api_methods.rs +++ b/crates/void-gmail/src/connector/api_methods.rs @@ -33,7 +33,8 @@ impl GmailConnector { debug!(config_id = %self.config_id, "token fresh, reusing"); } - Ok(GmailApiClient::new(&cache.access_token)) + Ok(GmailApiClient::new(&cache.access_token) + .with_store_limiter(&self.store_path, &self.config_id)) } pub async fn search_api( diff --git a/crates/void-gmail/src/connector/connector_trait.rs b/crates/void-gmail/src/connector/connector_trait.rs index 1126cd2..109b806 100644 --- a/crates/void-gmail/src/connector/connector_trait.rs +++ b/crates/void-gmail/src/connector/connector_trait.rs @@ -42,7 +42,8 @@ impl Connector for GmailConnector { let cache = auth::authorize_interactive(&creds, None).await?; cache.save(&token_path)?; - let api = GmailApiClient::new(&cache.access_token); + let api = GmailApiClient::new(&cache.access_token) + .with_store_limiter(&self.store_path, &self.config_id); let profile = api.get_profile().await?; info!( email = profile.email_address.as_deref().unwrap_or("?"), diff --git a/docs/connectors.md b/docs/connectors.md index 0b622ae..3fb3aae 100644 --- a/docs/connectors.md +++ b/docs/connectors.md @@ -55,6 +55,8 @@ Built-in OAuth2 credentials are included — **no Google Cloud setup required**: Gmail and Calendar share the same OAuth credentials, so adding the second one after the first is instant. By default Calendar syncs your primary calendar; list more with `calendar_ids`. +Gmail API calls from the CLI and the sync daemon share a per-account token bucket in the local store (~90 requests / minute) so they do not stampede after a quota error. + ## LinkedIn (Unipile) LinkedIn messages are synced through the [Unipile](https://www.unipile.com/) API. You need a Unipile account with a connected LinkedIn profile.