diff --git a/crates/taskito-core/src/resilience/rate_limiter.rs b/crates/taskito-core/src/resilience/rate_limiter.rs index dd57b8fc..34df21b2 100644 --- a/crates/taskito-core/src/resilience/rate_limiter.rs +++ b/crates/taskito-core/src/resilience/rate_limiter.rs @@ -118,6 +118,7 @@ mod tests { handles.push(std::thread::spawn(move || { barrier.wait(); // Retry on lock contention (SQLite busy) + let mut last_err = None; for _ in 0..10 { match limiter.try_acquire("concurrent_test", &config) { Ok(true) => { @@ -125,12 +126,14 @@ mod tests { return; } Ok(false) => return, - Err(_) => { + Err(e) => { + last_err = Some(e.to_string()); std::thread::sleep(std::time::Duration::from_millis(10)); continue; } } } + panic!("exhausted retries in SQLite rate-limit test: {last_err:?}"); })); } @@ -141,4 +144,79 @@ mod tests { // With 10 tokens and no refill, exactly 10 should succeed assert_eq!(acquired.load(Ordering::Relaxed), 10); } + + /// Postgres lost-update guard: under concurrency the row is seeded once and + /// read `FOR UPDATE`, so a fresh bucket must admit exactly `max_tokens` — not + /// more. Skips unless `TASKITO_POSTGRES_TEST_URL` is set (CI Postgres job). + #[cfg(feature = "postgres")] + #[test] + fn test_postgres_concurrent_token_acquisition_no_over_admit() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::{Arc, Barrier}; + + let url = match std::env::var("TASKITO_POSTGRES_TEST_URL") { + Ok(u) => u, + Err(_) => { + eprintln!("Skipping Postgres rate-limit test (TASKITO_POSTGRES_TEST_URL not set)"); + return; + } + }; + let num_threads = 12; + let storage = match crate::storage::postgres::PostgresStorage::with_pool_size( + &url, + num_threads as u32 + 4, + ) { + Ok(s) => StorageBackend::Postgres(s), + Err(e) => { + eprintln!("Skipping Postgres rate-limit test (cannot connect): {e}"); + return; + } + }; + let limiter = Arc::new(RateLimiter::new(storage)); + let config = RateLimitConfig { + max_tokens: 5.0, + refill_rate: 0.0, + }; + // Unique key so repeated runs against the same DB start from a full bucket. + let key = format!("pg-conc-{}", crate::job::now_millis()); + + let acquired = Arc::new(AtomicUsize::new(0)); + let barrier = Arc::new(Barrier::new(num_threads)); + let mut handles = vec![]; + for _ in 0..num_threads { + let limiter = limiter.clone(); + let config = config.clone(); + let acquired = acquired.clone(); + let barrier = barrier.clone(); + let key = key.clone(); + handles.push(std::thread::spawn(move || { + barrier.wait(); + let mut last_err = None; + for _ in 0..10 { + match limiter.try_acquire(&key, &config) { + Ok(true) => { + acquired.fetch_add(1, Ordering::Relaxed); + return; + } + Ok(false) => return, + Err(e) => { + last_err = Some(e.to_string()); + std::thread::sleep(std::time::Duration::from_millis(10)); + continue; + } + } + } + panic!("exhausted retries in Postgres rate-limit test: {last_err:?}"); + })); + } + for h in handles { + h.join().unwrap(); + } + + assert_eq!( + acquired.load(Ordering::Relaxed), + 5, + "FOR UPDATE + seed must admit exactly max_tokens, never over-admit" + ); + } } diff --git a/crates/taskito-core/src/storage/diesel_common/dead_letter.rs b/crates/taskito-core/src/storage/diesel_common/dead_letter.rs index 715e5f0a..19962019 100644 --- a/crates/taskito-core/src/storage/diesel_common/dead_letter.rs +++ b/crates/taskito-core/src/storage/diesel_common/dead_letter.rs @@ -18,7 +18,7 @@ macro_rules! impl_diesel_dead_letter_ops { // Write-priority transaction: this reads the job row then writes // it to `archived_jobs`, which would deadlock under SQLite's - // deferred lock-upgrade. See `archive_transaction`. + // deferred lock-upgrade. See `write_transaction`. let dlq_retry_count = job .metadata .as_deref() @@ -26,7 +26,7 @@ macro_rules! impl_diesel_dead_letter_ops { .and_then(|v| v.get("__dlq_retry_count")?.as_i64()) .unwrap_or(0) as i32; - self.archive_transaction(|conn| { + self.write_transaction(|conn| { let dlq_row = NewDeadLetterRow { id: &dlq_id, original_job_id: &job.id, diff --git a/crates/taskito-core/src/storage/diesel_common/jobs.rs b/crates/taskito-core/src/storage/diesel_common/jobs.rs index 16f161ce..62c718fe 100644 --- a/crates/taskito-core/src/storage/diesel_common/jobs.rs +++ b/crates/taskito-core/src/storage/diesel_common/jobs.rs @@ -164,9 +164,8 @@ macro_rules! impl_diesel_job_ops { pub fn enqueue(&self, new_job: NewJob) -> Result { let depends_on = new_job.depends_on.clone(); let job = new_job.into_job(); - let mut conn = self.conn()?; - conn.transaction(|conn| { + self.write_transaction(|conn| { // Validate dependencies exist and aren't dead/cancelled. // Terminal deps live in `archived_jobs`, so a missing live // row falls back to the archive. @@ -226,10 +225,12 @@ macro_rules! impl_diesel_job_ops { Ok(()) }) .map_err(|e| match e { - diesel::result::Error::RollbackTransaction => QueueError::DependencyNotFound( - "dependency not found or already dead/cancelled".to_string(), - ), - other => QueueError::Storage(other), + QueueError::Storage(diesel::result::Error::RollbackTransaction) => { + QueueError::DependencyNotFound( + "dependency not found or already dead/cancelled".to_string(), + ) + } + other => other, })?; Ok(job) @@ -243,10 +244,9 @@ macro_rules! impl_diesel_job_ops { // chunk size keeps the macro-generated code identical. const BATCH_INSERT_CHUNK: usize = 50; - let mut conn = self.conn()?; let jobs: Vec = new_jobs.into_iter().map(|nj| nj.into_job()).collect(); - conn.transaction(|conn| { + self.write_transaction(|conn| { let rows: Vec = jobs .iter() .map(|job| NewJobRow { @@ -304,7 +304,6 @@ macro_rules! impl_diesel_job_ops { pub fn enqueue_unique(&self, new_job: NewJob) -> Result { let depends_on = new_job.depends_on.clone(); let job = new_job.into_job(); - let mut conn = self.conn()?; // A UniqueViolation means a concurrent insert won the unique slot. // If that job is still active we return it; if it has since gone @@ -313,7 +312,7 @@ macro_rules! impl_diesel_job_ops { // an error instead of a phantom job that was never persisted. const MAX_ENQUEUE_ATTEMPTS: usize = 3; for _ in 0..MAX_ENQUEUE_ATTEMPTS { - let result = conn.transaction(|conn| { + let result = self.write_transaction(|conn| { // Return any existing active job with the same unique_key. if let Some(ref uk) = job.unique_key { let existing: Option = @@ -387,13 +386,17 @@ macro_rules! impl_diesel_job_ops { match result { Ok(j) => return Ok(j), - Err(diesel::result::Error::DatabaseError( + Err(QueueError::Storage(diesel::result::Error::DatabaseError( diesel::result::DatabaseErrorKind::UniqueViolation, _, - )) => { + ))) => { // Concurrent winner: return it if still active, else the // slot was freed by a terminal transition — retry insert. + // Acquire the connection here (not before the loop) so + // it never overlaps `write_transaction`'s own checkout — + // the in-memory test pool has a single connection. if let Some(ref uk) = job.unique_key { + let mut conn = self.conn()?; let existing: Option = jobs::table .filter(jobs::unique_key.eq(uk)) .filter(jobs::status.eq_any([ @@ -409,12 +412,12 @@ macro_rules! impl_diesel_job_ops { } continue; } - Err(diesel::result::Error::RollbackTransaction) => { + Err(QueueError::Storage(diesel::result::Error::RollbackTransaction)) => { return Err(QueueError::DependencyNotFound( "dependency not found or already dead/cancelled".to_string(), )); } - Err(e) => return Err(QueueError::Storage(e)), + Err(e) => return Err(e), } } @@ -432,9 +435,7 @@ macro_rules! impl_diesel_job_ops { now: i64, namespace: Option<&str>, ) -> Result> { - let mut conn = self.conn()?; - - conn.transaction(|conn| { + self.write_transaction(|conn| { let mut query = jobs::table .filter(jobs::queue.eq(queue_name)) .filter(jobs::status.eq(JobStatus::Pending as i32)) @@ -559,9 +560,7 @@ macro_rules! impl_diesel_job_ops { // to keep the loaded set small. let scan_limit = (max.saturating_mul(4)).min(400) as i64; - let mut conn = self.conn()?; - - conn.transaction(|conn| { + self.write_transaction(|conn| { let mut query = jobs::table .filter(jobs::queue.eq(queue_name)) .filter(jobs::status.eq(JobStatus::Pending as i32)) @@ -661,7 +660,7 @@ macro_rules! impl_diesel_job_ops { pub fn complete(&self, id: &str, result_bytes: Option>) -> Result<()> { let now = now_millis(); - self.archive_transaction(|conn| { + self.write_transaction(|conn| { let mut row: JobRow = match jobs::table .find(id) .filter(jobs::status.eq(JobStatus::Running as i32)) @@ -686,7 +685,7 @@ macro_rules! impl_diesel_job_ops { pub fn fail(&self, id: &str, error: &str) -> Result<()> { let now = now_millis(); - self.archive_transaction(|conn| { + self.write_transaction(|conn| { let mut row: JobRow = match jobs::table .find(id) .filter(jobs::status.eq(JobStatus::Running as i32)) @@ -758,7 +757,7 @@ macro_rules! impl_diesel_job_ops { pub fn cancel_job(&self, id: &str) -> Result { let now = now_millis(); - let archived = self.archive_transaction(|conn| { + let archived = self.write_transaction(|conn| { let mut row: JobRow = match jobs::table .find(id) .filter(jobs::status.eq(JobStatus::Pending as i32)) @@ -814,7 +813,7 @@ macro_rules! impl_diesel_job_ops { pub fn mark_cancelled(&self, id: &str) -> Result<()> { let now = now_millis(); - self.archive_transaction(|conn| { + self.write_transaction(|conn| { let mut row: JobRow = match jobs::table .find(id) .select(JobRow::as_select()) @@ -870,7 +869,7 @@ macro_rules! impl_diesel_job_ops { if !queue.is_empty() { let error_msg = format!("{reason}: {failed_job_id}"); - self.archive_transaction(|conn| { + self.write_transaction(|conn| { let rows: Vec = jobs::table .filter(jobs::id.eq_any(&queue)) .filter(jobs::status.eq(JobStatus::Pending as i32)) @@ -1283,7 +1282,7 @@ macro_rules! impl_diesel_job_ops { /// Purge completed jobs older than the given timestamp. Terminal /// jobs live in `archived_jobs`, so the purge targets that table. pub fn purge_completed(&self, older_than_ms: i64) -> Result { - self.archive_transaction(|conn| { + self.write_transaction(|conn| { let job_ids: Vec = archived_jobs::table .filter(archived_jobs::status.eq(JobStatus::Complete as i32)) .filter(archived_jobs::completed_at.lt(older_than_ms)) @@ -1306,7 +1305,7 @@ macro_rules! impl_diesel_job_ops { pub fn purge_completed_with_ttl(&self, global_cutoff_ms: i64) -> Result { let now = now_millis(); - self.archive_transaction(|conn| { + self.write_transaction(|conn| { let global_ids: Vec = archived_jobs::table .filter(archived_jobs::status.eq(JobStatus::Complete as i32)) .filter(archived_jobs::result_ttl_ms.is_null()) @@ -1418,7 +1417,7 @@ macro_rules! impl_diesel_job_ops { /// Expire pending jobs that have passed their expires_at. pub fn expire_pending_jobs(&self, now: i64) -> Result { - self.archive_transaction(|conn| { + self.write_transaction(|conn| { let rows: Vec = jobs::table .filter(jobs::status.eq(JobStatus::Pending as i32)) .filter(jobs::expires_at.is_not_null()) @@ -1434,7 +1433,7 @@ macro_rules! impl_diesel_job_ops { pub fn cancel_pending_by_queue(&self, queue: &str) -> Result { let now = now_millis(); - self.archive_transaction(|conn| { + self.write_transaction(|conn| { let rows: Vec = jobs::table .filter(jobs::status.eq(JobStatus::Pending as i32)) .filter(jobs::queue.eq(queue)) @@ -1449,7 +1448,7 @@ macro_rules! impl_diesel_job_ops { pub fn cancel_pending_by_task(&self, task_name: &str) -> Result { let now = now_millis(); - self.archive_transaction(|conn| { + self.write_transaction(|conn| { let rows: Vec = jobs::table .filter(jobs::status.eq(JobStatus::Pending as i32)) .filter(jobs::task_name.eq(task_name)) diff --git a/crates/taskito-core/src/storage/models.rs b/crates/taskito-core/src/storage/models.rs index 294cbba0..71a0260c 100644 --- a/crates/taskito-core/src/storage/models.rs +++ b/crates/taskito-core/src/storage/models.rs @@ -156,7 +156,17 @@ pub struct NewDeadLetterRow<'a> { } /// A row in the `rate_limits` table. -#[derive(Queryable, Selectable, Insertable, AsChangeset, Debug, Clone, Serialize, Deserialize)] +#[derive( + Queryable, + QueryableByName, + Selectable, + Insertable, + AsChangeset, + Debug, + Clone, + Serialize, + Deserialize, +)] #[diesel(table_name = rate_limits)] pub struct RateLimitRow { pub key: String, diff --git a/crates/taskito-core/src/storage/postgres/jobs.rs b/crates/taskito-core/src/storage/postgres/jobs.rs index a3bbcc76..c26f8bc2 100644 --- a/crates/taskito-core/src/storage/postgres/jobs.rs +++ b/crates/taskito-core/src/storage/postgres/jobs.rs @@ -16,9 +16,9 @@ crate::storage::diesel_common::impl_diesel_job_ops!(PostgresStorage, PgConnectio impl PostgresStorage { /// Run a read-then-write unit of work in a transaction. Postgres serializes /// row-level writes without the SQLite lock-upgrade deadlock, so a regular - /// transaction suffices; this mirrors [`SqliteStorage::archive_transaction`] + /// transaction suffices; this mirrors [`SqliteStorage::write_transaction`] /// so the shared job-ops macro can call one name on both backends. - pub(crate) fn archive_transaction(&self, f: F) -> Result + pub(crate) fn write_transaction(&self, f: F) -> Result where F: FnOnce(&mut PgConnection) -> std::result::Result, { diff --git a/crates/taskito-core/src/storage/postgres/mod.rs b/crates/taskito-core/src/storage/postgres/mod.rs index 4e64e839..2c616d8b 100644 --- a/crates/taskito-core/src/storage/postgres/mod.rs +++ b/crates/taskito-core/src/storage/postgres/mod.rs @@ -22,12 +22,13 @@ use diesel::r2d2::{ConnectionManager, Pool}; use crate::error::Result; use crate::storage::diesel_common::migrations as common_migrations; -/// Run an ALTER TABLE migration, logging a warning on any failure. -/// Postgres migrations use IF NOT EXISTS, so any error is genuinely unexpected. -fn migration_alter(conn: &mut PgConnection, sql: &str) { - if let Err(e) = diesel::sql_query(sql).execute(conn) { - log::warn!("migration failed for '{sql}': {e}"); - } +/// Run an ALTER TABLE migration. Postgres statements use `ADD COLUMN IF NOT +/// EXISTS`, so the already-applied case succeeds silently and any error is a +/// genuine failure (locked table, permissions, disk) that must be propagated +/// rather than leaving the schema missing a column. +fn migration_alter(conn: &mut PgConnection, sql: &str) -> Result<()> { + diesel::sql_query(sql).execute(conn)?; + Ok(()) } type PgPool = Pool>; @@ -151,7 +152,7 @@ impl PostgresStorage { diesel::sql_query(*sql).execute(&mut conn)?; } for sql in common_migrations::alter_statements(&common_migrations::POSTGRES) { - migration_alter(&mut conn, &sql); + migration_alter(&mut conn, &sql)?; } // Data backfills must fail loudly — a swallowed failure would leave // has_deps wrong and let dequeue bypass dependency enforcement. diff --git a/crates/taskito-core/src/storage/postgres/rate_limits.rs b/crates/taskito-core/src/storage/postgres/rate_limits.rs index 8e0abd9a..d10c7eac 100644 --- a/crates/taskito-core/src/storage/postgres/rate_limits.rs +++ b/crates/taskito-core/src/storage/postgres/rate_limits.rs @@ -35,27 +35,35 @@ impl PostgresStorage { /// Atomically try to acquire a rate limit token. /// Does the read-refill-consume-write in a single transaction to prevent /// race conditions between concurrent workers. + /// + /// The bucket row is seeded with `INSERT ... ON CONFLICT DO NOTHING` and then + /// read `FOR UPDATE`. A bare `SELECT ... FOR UPDATE` locks nothing when the + /// row is absent, so concurrent first-writers would each start from a full + /// bucket and over-admit; seeding first guarantees the row exists exactly once + /// (the unique-index conflict serializes the first writers) so the subsequent + /// `FOR UPDATE` always locks a committed row and the read-modify-write is + /// serialized. Mirrors the `FOR UPDATE` pattern in `acquire_lock`. pub fn try_acquire_token(&self, key: &str, max_tokens: f64, refill_rate: f64) -> Result { let mut conn = self.conn()?; let now = now_millis(); conn.transaction(|conn| { - let existing: Option = rate_limits::table - .find(key) - .select(RateLimitRow::as_select()) - .first(conn) - .optional()?; + diesel::sql_query( + "INSERT INTO rate_limits (key, tokens, max_tokens, refill_rate, last_refill) \ + VALUES ($1, $2, $2, $3, $4) ON CONFLICT (key) DO NOTHING", + ) + .bind::(key) + .bind::(max_tokens) + .bind::(refill_rate) + .bind::(now) + .execute(conn)?; - let mut row = match existing { - Some(r) => r, - None => RateLimitRow { - key: key.to_string(), - tokens: max_tokens, - max_tokens, - refill_rate, - last_refill: now, - }, - }; + let mut row: RateLimitRow = diesel::sql_query( + "SELECT key, tokens, max_tokens, refill_rate, last_refill \ + FROM rate_limits WHERE key = $1 FOR UPDATE", + ) + .bind::(key) + .get_result(conn)?; // Refill tokens based on elapsed time let elapsed_ms = (now - row.last_refill).max(0) as f64; diff --git a/crates/taskito-core/src/storage/sqlite/jobs.rs b/crates/taskito-core/src/storage/sqlite/jobs.rs index 5aa97725..01944635 100644 --- a/crates/taskito-core/src/storage/sqlite/jobs.rs +++ b/crates/taskito-core/src/storage/sqlite/jobs.rs @@ -16,13 +16,17 @@ crate::storage::diesel_common::impl_diesel_job_ops!(SqliteStorage, SqliteConnect impl SqliteStorage { /// Run a read-then-write unit of work under a write-priority transaction. /// - /// Archival reads a job row and then INSERTs/DELETEs it. Under SQLite a - /// plain deferred transaction starts as a reader and must later upgrade to - /// a writer; two such transactions racing each other deadlock and surface - /// as `SQLITE_BUSY` ("database is locked") without honoring `busy_timeout`. - /// `BEGIN IMMEDIATE` takes the write lock up front so the busy-timeout - /// backoff applies and the operations serialize cleanly. - pub(crate) fn archive_transaction(&self, f: F) -> Result + /// Any sequence that reads a row and then writes it (dequeue/claim, enqueue + /// with dependency validation, unique-key dedup, DLQ/archive moves, rate-limit + /// token consume) must use this. Under SQLite a plain deferred transaction + /// starts as a reader and must later upgrade to a writer; two such + /// transactions racing each other deadlock and surface as `SQLITE_BUSY` + /// ("database is locked") without honoring `busy_timeout`. `BEGIN IMMEDIATE` + /// takes the write lock up front so the busy-timeout backoff applies and the + /// operations serialize cleanly. The Postgres twin is a plain transaction + /// (MVCC has no deferred-upgrade hazard), so the name is uniform across + /// backends and callers never pick the wrong transaction kind. + pub(crate) fn write_transaction(&self, f: F) -> Result where F: FnOnce(&mut SqliteConnection) -> std::result::Result, { diff --git a/crates/taskito-core/src/storage/sqlite/mod.rs b/crates/taskito-core/src/storage/sqlite/mod.rs index 40dd838f..c5b3c71c 100644 --- a/crates/taskito-core/src/storage/sqlite/mod.rs +++ b/crates/taskito-core/src/storage/sqlite/mod.rs @@ -16,19 +16,19 @@ use diesel::prelude::*; use diesel::r2d2::{ConnectionManager, CustomizeConnection, Pool}; use diesel::sqlite::SqliteConnection; -use crate::error::Result; +use crate::error::{QueueError, Result}; use crate::storage::diesel_common::migrations as common_migrations; -/// Run an ALTER TABLE migration, suppressing "duplicate column" errors (SQLite). -fn migration_alter(conn: &mut SqliteConnection, sql: &str) { +/// Run an ALTER TABLE migration. SQLite has no `ADD COLUMN IF NOT EXISTS`, so a +/// "duplicate column" error means the column already exists and is ignored; +/// every other failure (locked DB, disk full, bad type) is propagated — a +/// silently-skipped ALTER would leave the schema missing a column and surface +/// later as a confusing query error far from the cause. +fn migration_alter(conn: &mut SqliteConnection, sql: &str) -> Result<()> { match diesel::sql_query(sql).execute(conn) { - Ok(_) => {} - Err(e) => { - let msg = e.to_string(); - if !msg.contains("duplicate column") { - log::warn!("migration failed for '{sql}': {e}"); - } - } + Ok(_) => Ok(()), + Err(e) if e.to_string().contains("duplicate column") => Ok(()), + Err(e) => Err(QueueError::Storage(e)), } } @@ -140,7 +140,7 @@ impl SqliteStorage { diesel::sql_query(*sql).execute(&mut conn)?; } for sql in common_migrations::alter_statements(&common_migrations::SQLITE) { - migration_alter(&mut conn, &sql); + migration_alter(&mut conn, &sql)?; } // Data backfills must fail loudly — a swallowed failure would leave // has_deps wrong and let dequeue bypass dependency enforcement. diff --git a/crates/taskito-core/src/storage/sqlite/rate_limits.rs b/crates/taskito-core/src/storage/sqlite/rate_limits.rs index 6b38c480..2ebf9ab1 100644 --- a/crates/taskito-core/src/storage/sqlite/rate_limits.rs +++ b/crates/taskito-core/src/storage/sqlite/rate_limits.rs @@ -30,13 +30,14 @@ impl SqliteStorage { } /// Atomically try to acquire a rate limit token. - /// Does the read-refill-consume-write in a single transaction to prevent - /// race conditions between concurrent workers. + /// Does the read-refill-consume-write in a single write transaction to + /// prevent race conditions between concurrent workers. Uses + /// [`SqliteStorage::write_transaction`] (BEGIN IMMEDIATE) so the read-then- + /// write can't hit the deferred-lock-upgrade `SQLITE_BUSY` deadlock. pub fn try_acquire_token(&self, key: &str, max_tokens: f64, refill_rate: f64) -> Result { - let mut conn = self.conn()?; let now = now_millis(); - conn.transaction(|conn| { + self.write_transaction(|conn| { let existing: Option = rate_limits::table .find(key) .select(RateLimitRow::as_select()) diff --git a/crates/taskito-core/tests/rust/storage_tests.rs b/crates/taskito-core/tests/rust/storage_tests.rs index 1cf76fd7..baea7540 100644 --- a/crates/taskito-core/tests/rust/storage_tests.rs +++ b/crates/taskito-core/tests/rust/storage_tests.rs @@ -655,6 +655,25 @@ fn run_storage_tests(s: &impl Storage) { test_dependent_blocked_by_cancelled_parent(s); test_payload_side_table(s); test_archived_job_payload_resolves(s); + test_rate_limit_token_exhaustion(s); +} + +fn test_rate_limit_token_exhaustion(s: &impl Storage) { + // With no refill, exactly `max_tokens` acquisitions succeed and the next + // fails. Locks the token-bucket contract on every backend (Postgres reads + // the row FOR UPDATE so this also guards the lost-update fix). + let key = "q-rate-exhaust"; + let max_tokens = 5.0; + for i in 0..5 { + assert!( + s.try_acquire_token(key, max_tokens, 0.0).unwrap(), + "token {i} should be granted" + ); + } + assert!( + !s.try_acquire_token(key, max_tokens, 0.0).unwrap(), + "bucket must be empty after max_tokens acquisitions" + ); } // ── Backend-specific wiring ──────────────────────────────────────────