feat(gmail): share a cross-process request token bucket

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
 

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<Duration, DbError> {
+        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)
     }

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;

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<Duration, DbError> {
+    if capacity <= 0.0 || refill_per_sec <= 0.0 {
+        return Ok(Duration::ZERO);
+    }
+
+    let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+    let raw: Option<String> = 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::<Bucket>(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)
+}

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;

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);
+}