diff --git a/README.md b/README.md
index efbade9..24965ea 100644
--- a/README.md
+++ b/README.md
@@ -39,7 +39,7 @@
 | `cachekitio` | ✅ | HTTP backend for [api.cachekit.io](https://api.cachekit.io) via [reqwest](https://crates.io/crates/reqwest) + rustls |
 | `encryption` | ✅ | Zero-knowledge AES-256-GCM via [cachekit-core](https://crates.io/crates/cachekit-core) |
 | `l1` | ✅ | In-process L1 cache via [moka](https://crates.io/crates/moka), with stale-while-revalidate (native) |
-| `reliability` | ✅ | Retry with backoff + jitter, circuit breaker, distributed fill locks (native only) |
+| `reliability` | ✅ | Retry with backoff + jitter, circuit breaker, backpressure, distributed fill locks (native only) |
 | `redis` | ❌ | Redis backend via [fred](https://crates.io/crates/fred) (native only) |
 | `memcached` | ❌ | Memcached backend via [rust-memcache](https://crates.io/crates/memcache) (native only) |
 | `file` | ❌ | Local filesystem backend, byte-compatible with cachekit-py's File backend (native only) |
@@ -391,11 +391,12 @@ With the `reliability` feature (default, native only), the `production`, `encryp
 |:------|:-------------|:---------|
 | **Retry** | Truncated exponential backoff + jitter on transient/timeout errors (`BackendErrorKind::is_retryable`); permanent and auth errors propagate immediately | 3 attempts, 100 ms base, 5 s cap, jitter ×[0.5, 1.5) |
 | **Circuit breaker** | closed → open after N retryable failures in a rolling window; fails fast (`BackendErrorKind::CircuitOpen`) while open; half-open probes recovery | threshold 5, window 60 s, open 5 s, 3 probes, close after 3 successes |
-| **Graceful degradation** | On outage-class backend failure (transient, timeout, open breaker), `#[cachekit]`-wrapped functions run uncached (fail-open); permanent/auth errors propagate — a wrong API key fails loudly. `secure` paths fail **closed** on everything — encrypted workloads never silently degrade | built into the macro |
+| **Backpressure** | Bounds concurrent backend data ops with a semaphore + bounded waiting queue; over-limit calls are shed with `BackendErrorKind::Backpressure` before reaching the backend — a slow backend can't exhaust the caller's connection pool or memory | 100 concurrent, 1 000 queued, 100 ms wait (Python SDK parity) |
+| **Graceful degradation** | On outage-class backend failure (transient, timeout, open breaker, backpressure shed), `#[cachekit]`-wrapped functions run uncached (fail-open); permanent/auth errors propagate — a wrong API key fails loudly. `secure` paths fail **closed** on everything — encrypted workloads never silently degrade | built into the macro |
 | **Single-flight** | Concurrent misses of one key collapse to a single execution: per-key in-process lock, plus a distributed fill lock across processes on lock-capable backends (cachekit.io, Redis) | in-process always on; cross-process 5 s lock, 100 ms polls |
 | **Stale-while-revalidate** | Stale-but-unexpired L1 hits are served immediately while one single-flight-deduplicated background task re-executes the function ([details](#stale-while-revalidate-swr)) | on by default with `l1` (native); threshold 0.5 × entry TTL ±10% jitter |
 
-Retry sits *inside* the breaker (one exhausted retry sequence = one breaker failure), degradation, single-flight, and SWR sit in the `#[cachekit]` macro around the read path — the same composition as the TypeScript SDK's `ReliabilityExecutor` and the Python decorator.
+Retry sits *inside* the breaker (one exhausted retry sequence = one breaker failure) and backpressure sits *outside* both — one permit per logical operation, held across the whole retry sequence, so retry amplification is bounded and shed calls never skew breaker state. Degradation, single-flight, and SWR sit in the `#[cachekit]` macro around the read path — the same composition as the TypeScript SDK's `ReliabilityExecutor` and the Python decorator.
 
 ```rust,ignore
 use std::time::Duration;
@@ -409,9 +410,9 @@ let cache = CacheKit::production("redis://localhost:6379").await?
     })
     .build()?;
 
-// Opt a preset out: a config with both layers `None` applies no wrapping.
+// Opt a preset out: a config with all layers `None` applies no wrapping.
 let bare = CacheKit::production("redis://localhost:6379").await?
-    .reliability(ReliabilityConfig { retry: None, circuit_breaker: None })
+    .reliability(ReliabilityConfig { retry: None, circuit_breaker: None, backpressure: None })
     .build()?;
 ```
 
diff --git a/crates/cachekit-macros/src/lib.rs b/crates/cachekit-macros/src/lib.rs
index 9b0ec74..3195f5e 100644
--- a/crates/cachekit-macros/src/lib.rs
+++ b/crates/cachekit-macros/src/lib.rs
@@ -228,8 +228,9 @@ fn extract_ok_type(ret: &ReturnType) -> syn::Result<Type> {
 ///   must be `Send`. Refresh failures are absorbed: the stale value keeps
 ///   serving until hard expiry, where the blocking path surfaces errors.
 /// - **Graceful degradation**: on an outage-class backend failure —
-///   transient, timeout, or an open circuit breaker — the plain path fails
-///   *open*: the function executes uncached and its result is returned.
+///   transient, timeout, an open circuit breaker, or a backpressure shed —
+///   the plain path fails *open*: the function executes uncached and its
+///   result is returned.
 ///   Permanent and authentication errors propagate even on the plain path
 ///   (a wrong API key must fail loudly, not silently disable caching
 ///   forever). With `secure`, *every* backend and decryption error fails
@@ -422,14 +423,19 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result<TokenStream2> {
     }
 
     // Graceful degradation (LAB-518): on an OUTAGE-class backend failure —
-    // retryable (transient/timeout) or a fast-failing open circuit breaker —
-    // the plain path fails OPEN: the wrapped function runs uncached, so a
-    // cache outage costs performance, not availability. Permanent and
-    // authentication errors PROPAGATE even on the plain path: a wrong API
-    // key that silently fell open would run uncached forever with zero
-    // signal while looking healthy (expert-panel finding). The `secure`
-    // path stays fail-CLOSED on everything: backend and decryption errors
-    // reach the caller, so an encrypted workload never silently degrades.
+    // retryable (transient/timeout), a fast-failing open circuit breaker,
+    // or a backpressure shed (LAB-729) — the plain path fails OPEN: the
+    // wrapped function runs uncached, so a cache outage costs performance,
+    // not availability. A shed is the same class as CircuitOpen (the call
+    // never reached the backend), and failing open adds no work a cold-miss
+    // storm doesn't already create: the limiter bounds backend cache ops,
+    // not origin executions, and single-flight still dedupes those.
+    // Permanent and authentication errors PROPAGATE even on the plain path:
+    // a wrong API key that silently fell open would run uncached forever
+    // with zero signal while looking healthy (expert-panel finding). The
+    // `secure` path stays fail-CLOSED on everything: backend and decryption
+    // errors reach the caller, so an encrypted workload never silently
+    // degrades.
     let fail_open_arm = if args.secure {
         quote! {}
     } else {
@@ -439,6 +445,7 @@ fn expand(args: &MacroArgs, mut func: ItemFn) -> syn::Result<TokenStream2> {
                     || matches!(
                         __ck_be.kind,
                         cachekit::error::BackendErrorKind::CircuitOpen
+                            | cachekit::error::BackendErrorKind::Backpressure
                     ) => {}
         }
     };
diff --git a/crates/cachekit/src/client.rs b/crates/cachekit/src/client.rs
index 3b7383c..634dd61 100644
--- a/crates/cachekit/src/client.rs
+++ b/crates/cachekit/src/client.rs
@@ -975,11 +975,12 @@ impl CacheKitBuilder {
     }
 
     /// Wrap the backend in the reliability stack (retry with exponential
-    /// backoff + jitter, circuit breaker) — see [`crate::reliability`].
+    /// backoff + jitter, circuit breaker, backpressure) — see
+    /// [`crate::reliability`].
     ///
     /// Enabled by default with production settings by the `production`,
     /// `encrypted`, and `io` intent presets; off for `minimal` and for
-    /// manually-built clients. To opt a preset out, pass a config with both
+    /// manually-built clients. To opt a preset out, pass a config with all
     /// layers `None` — an empty config applies no wrapping at all.
     #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
     pub fn reliability(mut self, config: crate::reliability::ReliabilityConfig) -> Self {
@@ -1073,11 +1074,15 @@ impl CacheKitBuilder {
         };
 
         // Apply the reliability stack last so it decorates the final backend.
-        // A config with neither layer set is the documented opt-out: skip the
+        // A config with no layer set is the documented opt-out: skip the
         // (no-op) decorator entirely.
         #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
         let backend = match self.reliability {
-            Some(config) if config.retry.is_some() || config.circuit_breaker.is_some() => {
+            Some(config)
+                if config.retry.is_some()
+                    || config.circuit_breaker.is_some()
+                    || config.backpressure.is_some() =>
+            {
                 crate::reliability::wrap_reliable(backend, config)
             }
             _ => backend,
diff --git a/crates/cachekit/src/error.rs b/crates/cachekit/src/error.rs
index 8a4bd93..9bc1acd 100644
--- a/crates/cachekit/src/error.rs
+++ b/crates/cachekit/src/error.rs
@@ -52,6 +52,11 @@ pub enum BackendErrorKind {
     /// the backend. Not retryable *now*; the breaker re-probes on its own
     /// schedule (see `reliability::CircuitBreakerConfig::open_timeout`).
     CircuitOpen,
+    /// The concurrency limiter shed this call before it reached the backend:
+    /// the waiting queue was full, or no permit freed up within the acquire
+    /// timeout. Not retryable *now* — an immediate retry would re-join the
+    /// same overloaded queue (see `reliability::BackpressureConfig`).
+    Backpressure,
 }
 
 impl BackendErrorKind {
@@ -70,6 +75,7 @@ impl std::fmt::Display for BackendErrorKind {
             Self::Timeout => write!(f, "timeout"),
             Self::Authentication => write!(f, "authentication"),
             Self::CircuitOpen => write!(f, "circuit-open"),
+            Self::Backpressure => write!(f, "backpressure"),
         }
     }
 }
@@ -140,6 +146,16 @@ impl BackendError {
         }
     }
 
+    /// Create a backpressure backend error (call shed by the concurrency
+    /// limiter, backend not reached).
+    pub fn backpressure(message: impl Into<String>) -> Self {
+        Self {
+            kind: BackendErrorKind::Backpressure,
+            message: message.into(),
+            source: None,
+        }
+    }
+
     /// Sanitize error messages to strip API keys (CWE-532).
     pub fn sanitize_message(msg: &str, api_key: &str) -> String {
         if api_key.is_empty() {
diff --git a/crates/cachekit/src/intents.rs b/crates/cachekit/src/intents.rs
index 94d916b..1008f8f 100644
--- a/crates/cachekit/src/intents.rs
+++ b/crates/cachekit/src/intents.rs
@@ -12,10 +12,11 @@
 //! | `encrypted` | Redis | On | AES-256-GCM | Yes | On | 600 s |
 //! | `io` | cachekit.io | On | No | n/a (HTTP) | On | 3 600 s |
 //!
-//! ¹ Retry with backoff + jitter and a circuit breaker around backend ops
-//! (requires the `reliability` feature, on by default — see
-//! [`crate::reliability`]). Override via [`CacheKitBuilder::reliability`];
-//! a config with both layers `None` disables the stack entirely.
+//! ¹ Retry with backoff + jitter, a circuit breaker, and backpressure
+//! (bounded backend concurrency) around backend ops (requires the
+//! `reliability` feature, on by default — see [`crate::reliability`]).
+//! Override via [`CacheKitBuilder::reliability`]; a config with all layers
+//! `None` disables the stack entirely.
 
 use std::time::Duration;
 
@@ -43,8 +44,9 @@ impl CacheKit {
     ///   connection is not re-established)
     /// * L1 cache: **off**
     /// * Encryption: **no**
-    /// * Reliability: **off** — no retry, no circuit breaker; every backend
-    ///   error propagates on first failure
+    /// * Reliability: **off** — no retry, no circuit breaker, no
+    ///   backpressure; every backend error propagates on first failure and
+    ///   backend concurrency is unbounded
     /// * Default TTL: **300 s**
     ///
     /// Good for: product catalogs, public data, development.
@@ -82,7 +84,8 @@ impl CacheKit {
     ///   exponential backoff after a dropped connection)
     /// * L1 cache: **on** (1 000 entries)
     /// * Encryption: **no**
-    /// * Reliability: **on** — retry with backoff + jitter, circuit breaker
+    /// * Reliability: **on** — retry with backoff + jitter, circuit
+    ///   breaker, backpressure (max 100 concurrent backend ops)
     /// * Default TTL: **600 s**
     ///
     /// Good for: user sessions, API responses, production services.
@@ -124,7 +127,8 @@ impl CacheKit {
     ///   exponential backoff after a dropped connection)
     /// * L1 cache: **on** (1 000 entries, stores ciphertext)
     /// * Encryption: **AES-256-GCM** with HKDF-SHA256
-    /// * Reliability: **on** — retry with backoff + jitter, circuit breaker
+    /// * Reliability: **on** — retry with backoff + jitter, circuit
+    ///   breaker, backpressure (max 100 concurrent backend ops)
     /// * Default TTL: **600 s**
     /// * Tenant ID: `"default"` (override via
     ///   [`.encryption_from_bytes()`](CacheKitBuilder::encryption_from_bytes))
@@ -178,7 +182,8 @@ impl CacheKit {
     /// * L1 cache: **on** (1 000 entries)
     /// * Encryption: **no** (add via
     ///   [`.encryption()`](CacheKitBuilder::encryption))
-    /// * Reliability: **on** — retry with backoff + jitter, circuit breaker
+    /// * Reliability: **on** — retry with backoff + jitter, circuit
+    ///   breaker, backpressure (max 100 concurrent backend ops)
     /// * Default TTL: **3 600 s**
     ///
     /// Good for: serverless, edge compute, managed caching without Redis.
diff --git a/crates/cachekit/src/lib.rs b/crates/cachekit/src/lib.rs
index f78572c..27c4f38 100644
--- a/crates/cachekit/src/lib.rs
+++ b/crates/cachekit/src/lib.rs
@@ -81,7 +81,7 @@ pub use cachekit_macros::cachekit;
 pub use flight::SingleFlight;
 
 #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
-pub use reliability::{CircuitBreakerConfig, ReliabilityConfig, RetryConfig};
+pub use reliability::{BackpressureConfig, CircuitBreakerConfig, ReliabilityConfig, RetryConfig};
 
 // ── Shared jitter source ─────────────────────────────────────────────────────
 
diff --git a/crates/cachekit/src/reliability.rs b/crates/cachekit/src/reliability.rs
index 6d61751..51bc89b 100644
--- a/crates/cachekit/src/reliability.rs
+++ b/crates/cachekit/src/reliability.rs
@@ -1,16 +1,28 @@
-//! Reliability tier: retry with exponential backoff + jitter, and a
-//! closed/open/half-open circuit breaker around backend operations.
+//! Reliability tier: backpressure (bounded backend concurrency), retry with
+//! exponential backoff + jitter, and a closed/open/half-open circuit breaker
+//! around backend operations.
 //!
-//! Both are composed by a private `ReliableBackend` decorator around any
+//! All three are composed by a private `ReliableBackend` decorator around any
 //! [`crate::backend::Backend`], applied by the builder when a [`ReliabilityConfig`] is set
 //! (see [`crate::CacheKitBuilder::reliability`]). The intent presets
 //! `production`, `encrypted`, and `io` enable it by default; `minimal` does
 //! not — mirroring the TypeScript SDK's preset posture.
 //!
-//! Composition order matches the TypeScript SDK's `ReliabilityExecutor`:
-//! the retry loop is *inside* the breaker, so one exhausted retry sequence
-//! counts as a single breaker failure, and a fast-failing open breaker never
-//! spends time retrying.
+//! Composition order is `backpressure(breaker(retry(op)))`:
+//!
+//! - The retry loop is *inside* the breaker (matching the TypeScript SDK's
+//!   `ReliabilityExecutor`), so one exhausted retry sequence counts as a
+//!   single breaker failure, and a fast-failing open breaker never spends
+//!   time retrying.
+//! - The concurrency limiter is *outermost*: one permit per logical cache
+//!   operation, held across the entire breaker/retry sequence. That bounds
+//!   in-flight work including retry amplification (K callers mid-backoff are
+//!   still K permits — new work queues behind them instead of piling onto a
+//!   struggling backend), and a shed call never touches breaker counters or
+//!   half-open probe slots, so the breaker keeps measuring backend health,
+//!   not caller-side overload. A permit holder never re-enters the limiter
+//!   (backend ops don't nest), so holding permits across retry backoff
+//!   cannot deadlock.
 //!
 //! Unlike the TypeScript breaker (which counts every error), only errors
 //! classified retryable by [`crate::error::BackendErrorKind::is_retryable`] (`Transient`,
@@ -90,9 +102,50 @@ impl Default for CircuitBreakerConfig {
     }
 }
 
+/// Backpressure configuration: bound how many backend data operations may be
+/// in flight at once, so a slow or failing backend cannot exhaust the
+/// caller's connection pool or memory.
+///
+/// Defaults mirror the Python SDK's `BackpressureConfig`
+/// (`max_concurrent_requests: 100`, `queue_size: 1000`, `timeout: 0.1s`).
+///
+/// Over-limit calls first join a bounded waiting queue; a caller that finds
+/// the queue full, or that waits longer than [`Self::acquire_timeout`]
+/// without a permit freeing up, is shed with a
+/// [`crate::error::BackendErrorKind::Backpressure`] error — never queued
+/// unboundedly. Shed calls do not reach the backend and do not count toward
+/// opening the circuit breaker.
+///
+/// On the `#[cachekit]` macro's plain path a shed is outage-class — exactly
+/// like `CircuitOpen`, the wrapped function runs uncached (fail-open);
+/// `secure` paths fail closed on a shed like on every other backend error.
+#[derive(Debug, Clone, PartialEq)]
+pub struct BackpressureConfig {
+    /// Maximum backend data operations in flight at once (default: 100).
+    /// `0` behaves as `1`.
+    pub max_concurrent: usize,
+    /// Maximum callers waiting for a permit before further calls are shed
+    /// immediately (default: 1000). `0` disables waiting entirely: a call
+    /// that cannot take a permit on the spot is shed.
+    pub max_queue: usize,
+    /// How long a queued caller waits for a permit before it is shed
+    /// (default: 100 ms).
+    pub acquire_timeout: Duration,
+}
+
+impl Default for BackpressureConfig {
+    fn default() -> Self {
+        Self {
+            max_concurrent: 100,
+            max_queue: 1000,
+            acquire_timeout: Duration::from_millis(100),
+        }
+    }
+}
+
 /// Reliability stack configuration: which layers to apply around backend ops.
 ///
-/// The `Default` enables both layers with production defaults. Disable a
+/// The `Default` enables all layers with production defaults. Disable a
 /// layer by setting its field to `None`:
 ///
 /// ```
@@ -100,6 +153,7 @@ impl Default for CircuitBreakerConfig {
 ///
 /// let retry_only = ReliabilityConfig {
 ///     circuit_breaker: None,
+///     backpressure: None,
 ///     ..ReliabilityConfig::default()
 /// };
 /// assert!(retry_only.retry.is_some());
@@ -110,6 +164,8 @@ pub struct ReliabilityConfig {
     pub retry: Option<RetryConfig>,
     /// Circuit breaker, or `None` to never fail fast.
     pub circuit_breaker: Option<CircuitBreakerConfig>,
+    /// Concurrency limiter, or `None` for unbounded backend concurrency.
+    pub backpressure: Option<BackpressureConfig>,
 }
 
 impl Default for ReliabilityConfig {
@@ -117,6 +173,7 @@ impl Default for ReliabilityConfig {
         Self {
             retry: Some(RetryConfig::default()),
             circuit_breaker: Some(CircuitBreakerConfig::default()),
+            backpressure: Some(BackpressureConfig::default()),
         }
     }
 }
@@ -394,21 +451,101 @@ impl Drop for ProbePermit<'_> {
     }
 }
 
+// ── ConcurrencyLimiter ───────────────────────────────────────────────────────
+
+/// Bounds concurrent backend data operations with a semaphore and a bounded
+/// waiting queue (two-phase, like the Python SDK's `BackpressureController`):
+/// a saturated limiter admits up to `max_queue` waiters for at most
+/// `acquire_timeout` each; everyone else is shed with a
+/// [`crate::error::BackendErrorKind::Backpressure`] error.
+#[derive(Debug)]
+pub(crate) struct ConcurrencyLimiter {
+    semaphore: tokio::sync::Semaphore,
+    /// Callers currently waiting for a permit (phase-2 queue depth).
+    waiting: std::sync::atomic::AtomicUsize,
+    config: BackpressureConfig,
+}
+
+/// RAII guard for a slot in the waiting queue: decrements `waiting` on every
+/// exit path, including cancellation mid-`acquire` (same lesson as
+/// [`ProbePermit`] — manual increment/decrement pairs leak on cancel).
+struct QueueSlot<'a> {
+    waiting: &'a std::sync::atomic::AtomicUsize,
+}
+
+impl Drop for QueueSlot<'_> {
+    fn drop(&mut self) {
+        self.waiting
+            .fetch_sub(1, std::sync::atomic::Ordering::AcqRel);
+    }
+}
+
+impl ConcurrencyLimiter {
+    pub(crate) fn new(config: BackpressureConfig) -> Self {
+        Self {
+            // `Semaphore::new(0)` would shed every call after acquire_timeout
+            // with nothing ever admitted — clamp like RetryConfig's "0
+            // behaves as 1".
+            semaphore: tokio::sync::Semaphore::new(config.max_concurrent.max(1)),
+            waiting: std::sync::atomic::AtomicUsize::new(0),
+            config,
+        }
+    }
+
+    /// Take a permit, or shed the call.
+    ///
+    /// Phase 1: a free permit is taken on the spot — no queue accounting.
+    /// Phase 2 (saturated): join the bounded waiting queue and wait up to
+    /// `acquire_timeout` for a permit; queue-full and wait-timeout both shed
+    /// with a `Backpressure` error. The returned permit releases on drop, so
+    /// a cancelled or panicking caller can never leak capacity.
+    async fn acquire(&self) -> Result<tokio::sync::SemaphorePermit<'_>, BackendError> {
+        use std::sync::atomic::Ordering;
+
+        if let Ok(permit) = self.semaphore.try_acquire() {
+            return Ok(permit);
+        }
+        if self.waiting.fetch_add(1, Ordering::AcqRel) >= self.config.max_queue {
+            self.waiting.fetch_sub(1, Ordering::AcqRel);
+            return Err(BackendError::backpressure(format!(
+                "backpressure: waiting queue is full (max_queue={}), call shed without reaching the backend",
+                self.config.max_queue
+            )));
+        }
+        let _slot = QueueSlot {
+            waiting: &self.waiting,
+        };
+        match tokio::time::timeout(self.config.acquire_timeout, self.semaphore.acquire()).await {
+            Ok(Ok(permit)) => Ok(permit),
+            // The semaphore is never closed; treat a close defensively as shed.
+            Ok(Err(_closed)) => Err(BackendError::backpressure(
+                "backpressure: limiter unavailable, call shed without reaching the backend",
+            )),
+            Err(_elapsed) => Err(BackendError::backpressure(format!(
+                "backpressure: timed out waiting for a permit after {:?}, call shed without reaching the backend",
+                self.config.acquire_timeout
+            ))),
+        }
+    }
+}
+
 // ── ReliableBackend ──────────────────────────────────────────────────────────
 
 /// Decorator that applies the reliability stack to every cache operation of
-/// an inner [`Backend`]: `breaker(retry(op))`.
+/// an inner [`Backend`]: `backpressure(breaker(retry(op)))`.
 ///
-/// - `get`/`set`/`delete`/`exists` are retried on retryable errors and gated
-///   by the circuit breaker.
+/// - `get`/`set`/`delete`/`exists` take a concurrency-limiter permit, are
+///   retried on retryable errors, and gated by the circuit breaker.
 /// - `health` passes through unguarded — it is a diagnostic and must keep
-///   reporting truthfully while the breaker fails data calls fast.
+///   reporting truthfully while the breaker fails data calls fast (or the
+///   limiter sheds them).
 /// - [`Backend::as_lockable`] forwards to the inner backend so distributed
-///   fill locks bypass retry/breaker (locks are best-effort advisory).
+///   fill locks bypass the stack (locks are best-effort advisory).
 pub(crate) struct ReliableBackend {
     inner: SharedBackend,
     retry: Option<RetryPolicy>,
     breaker: Option<CircuitBreaker>,
+    limiter: Option<ConcurrencyLimiter>,
 }
 
 impl ReliableBackend {
@@ -417,6 +554,7 @@ impl ReliableBackend {
             inner,
             retry: config.retry.map(RetryPolicy::new),
             breaker: config.circuit_breaker.map(CircuitBreaker::new),
+            limiter: config.backpressure.map(ConcurrencyLimiter::new),
         }
     }
 
@@ -425,6 +563,13 @@ impl ReliableBackend {
         F: Fn() -> Fut,
         Fut: Future<Output = Result<T, BackendError>>,
     {
+        // Outermost layer: one permit per logical operation, held across the
+        // whole breaker/retry sequence (see the module docs for why). A shed
+        // call returns here — before touching breaker state.
+        let _permit = match &self.limiter {
+            Some(limiter) => Some(limiter.acquire().await?),
+            None => None,
+        };
         let permit = match &self.breaker {
             Some(cb) => Some(cb.try_acquire()?),
             None => None,
@@ -643,6 +788,108 @@ mod tests {
         assert_eq!(cb.state(), CircuitState::Closed);
     }
 
+    #[test]
+    fn reliability_default_enables_backpressure_with_python_parity_defaults() {
+        let config = ReliabilityConfig::default();
+        let bp = config.backpressure.expect("backpressure is on by default");
+        assert_eq!(bp.max_concurrent, 100);
+        assert_eq!(bp.max_queue, 1000);
+        assert_eq!(bp.acquire_timeout, Duration::from_millis(100));
+    }
+
+    #[tokio::test]
+    async fn limiter_clamps_zero_max_concurrent_to_one() {
+        let limiter = ConcurrencyLimiter::new(BackpressureConfig {
+            max_concurrent: 0,
+            max_queue: 0,
+            acquire_timeout: Duration::from_millis(10),
+        });
+        let permit = limiter
+            .acquire()
+            .await
+            .expect("0 behaves as 1 — one permit exists");
+        drop(permit);
+    }
+
+    #[tokio::test]
+    async fn limiter_sheds_immediately_when_queue_disabled() {
+        let limiter = ConcurrencyLimiter::new(BackpressureConfig {
+            max_concurrent: 1,
+            max_queue: 0,
+            acquire_timeout: Duration::from_secs(5),
+        });
+        let _held = limiter.acquire().await.expect("first permit");
+        let start = Instant::now();
+        let err = limiter
+            .acquire()
+            .await
+            .expect_err("saturated with no waiting queue");
+        assert_eq!(err.kind, BackendErrorKind::Backpressure);
+        assert!(!err.kind.is_retryable());
+        assert!(
+            start.elapsed() < Duration::from_millis(500),
+            "queue-full sheds immediately, not after acquire_timeout"
+        );
+    }
+
+    #[tokio::test]
+    async fn limiter_waiting_slot_released_on_cancelled_wait() {
+        // A waiter cancelled mid-acquire (caller timeout / select!) must free
+        // its queue slot via the QueueSlot drop guard. With max_queue: 1, a
+        // leaked slot would shed the next waiter instantly as queue-full;
+        // joining the queue (observable as waiting out the acquire_timeout)
+        // proves the slot was released.
+        let limiter = ConcurrencyLimiter::new(BackpressureConfig {
+            max_concurrent: 1,
+            max_queue: 1,
+            acquire_timeout: Duration::from_millis(100),
+        });
+        let _held = limiter.acquire().await.expect("first permit");
+        let cancelled = tokio::time::timeout(Duration::from_millis(20), limiter.acquire()).await;
+        assert!(cancelled.is_err(), "waiter cancelled from outside");
+
+        let start = Instant::now();
+        let err = limiter
+            .acquire()
+            .await
+            .expect_err("permit never frees, waiter times out");
+        assert_eq!(err.kind, BackendErrorKind::Backpressure);
+        assert!(
+            start.elapsed() >= Duration::from_millis(80),
+            "must join the queue and wait out acquire_timeout — an instant \
+             queue-full shed means the cancelled waiter leaked its slot"
+        );
+    }
+
+    #[tokio::test]
+    async fn limiter_sheds_queue_full_at_nonzero_boundary() {
+        // cap 1, queue 1: with the permit held and one waiter parked, a
+        // third caller must shed instantly as queue-full — pinning the
+        // fetch_add boundary arithmetic at a nonzero max_queue.
+        let limiter = ConcurrencyLimiter::new(BackpressureConfig {
+            max_concurrent: 1,
+            max_queue: 1,
+            acquire_timeout: Duration::from_millis(200),
+        });
+        let _held = limiter.acquire().await.expect("first permit");
+        let waiter = async {
+            // Parks in the queue immediately and times out after 200 ms.
+            limiter.acquire().await
+        };
+        let third = async {
+            tokio::time::sleep(Duration::from_millis(50)).await; // waiter parked
+            let start = Instant::now();
+            let err = limiter.acquire().await.expect_err("queue of 1 is full");
+            assert_eq!(err.kind, BackendErrorKind::Backpressure);
+            assert!(
+                start.elapsed() < Duration::from_millis(100),
+                "queue-full sheds instantly, not after the wait timeout"
+            );
+        };
+        let (waited, ()) = tokio::join!(waiter, third);
+        waited.expect_err("the parked waiter itself times out");
+    }
+
     #[test]
     fn retry_delay_is_capped_and_jittered() {
         let policy = RetryPolicy::new(RetryConfig {
diff --git a/crates/cachekit/tests/macro_tests.rs b/crates/cachekit/tests/macro_tests.rs
index 16f2706..c3821e3 100644
--- a/crates/cachekit/tests/macro_tests.rs
+++ b/crates/cachekit/tests/macro_tests.rs
@@ -565,6 +565,7 @@ async fn macro_fails_open_when_circuit_is_open() {
                 open_timeout: Duration::from_secs(60),
                 ..CircuitBreakerConfig::default()
             }),
+            backpressure: None,
         })
         .no_l1()
         .build()
@@ -599,3 +600,145 @@ async fn macro_fails_open_when_circuit_is_open() {
         "both outage classes (transient, circuit-open) fall open to the body"
     );
 }
+
+// ── Backpressure fail-open (LAB-729) ─────────────────────────────────────────
+
+/// Backend whose `get` never completes — parks a caller on the single
+/// backpressure permit so every subsequent data op is shed.
+#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
+#[derive(Debug, Default)]
+struct HangingInner {
+    calls: std::sync::atomic::AtomicU32,
+}
+
+#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
+#[derive(Debug, Default, Clone)]
+struct HangingBackend {
+    inner: std::sync::Arc<HangingInner>,
+}
+
+#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
+impl HangingBackend {
+    fn new_with_handle() -> (SharedBackend, Self) {
+        let backend = Self::default();
+        let handle = backend.clone();
+        #[cfg(not(feature = "unsync"))]
+        let shared: SharedBackend = std::sync::Arc::new(backend);
+        #[cfg(feature = "unsync")]
+        let shared: SharedBackend = std::rc::Rc::new(backend);
+        (shared, handle)
+    }
+}
+
+#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
+#[cfg_attr(not(feature = "unsync"), async_trait)]
+#[cfg_attr(feature = "unsync", async_trait(?Send))]
+impl Backend for HangingBackend {
+    async fn get(&self, _key: &str) -> Result<Option<Vec<u8>>, BackendError> {
+        self.inner
+            .calls
+            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        std::future::pending().await
+    }
+
+    async fn set(
+        &self,
+        _key: &str,
+        _value: Vec<u8>,
+        _ttl: Option<Duration>,
+    ) -> Result<(), BackendError> {
+        Ok(())
+    }
+
+    async fn delete(&self, _key: &str) -> Result<bool, BackendError> {
+        Ok(false)
+    }
+
+    async fn exists(&self, _key: &str) -> Result<bool, BackendError> {
+        Ok(false)
+    }
+
+    async fn health(&self) -> Result<HealthStatus, BackendError> {
+        Ok(HealthStatus {
+            is_healthy: true,
+            latency_ms: 0.0,
+            backend_type: "hanging".to_owned(),
+            details: HashMap::new(),
+        })
+    }
+}
+
+#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
+static SHED_RUNS: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
+
+#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
+#[cachekit(client = cache, ttl = 60, interop = "shed_op", namespace = "reliab")]
+async fn shed_op(cache: &CacheKit, id: u64) -> Result<User, CachekitError> {
+    SHED_RUNS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+    Ok(User {
+        name: format!("shed {id}"),
+    })
+}
+
+/// A backpressure shed is outage-class (LAB-729): like `CircuitOpen`, the
+/// call never reached the backend, so the plain path must run the body
+/// uncached instead of surfacing the `Backpressure` error.
+#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
+#[tokio::test]
+async fn macro_fails_open_when_backpressure_sheds() {
+    use cachekit::reliability::{BackpressureConfig, ReliabilityConfig};
+
+    let (shared, handle) = HangingBackend::new_with_handle();
+    let cache = std::sync::Arc::new(
+        CacheKit::builder()
+            .backend(shared)
+            .reliability(ReliabilityConfig {
+                retry: None,
+                circuit_breaker: None,
+                backpressure: Some(BackpressureConfig {
+                    max_concurrent: 1,
+                    max_queue: 0,
+                    acquire_timeout: Duration::from_secs(5),
+                }),
+            })
+            .no_l1()
+            .build()
+            .expect("client builds"),
+    );
+
+    // Park one caller on the never-completing backend: it holds the single
+    // permit for the duration of the test.
+    let holder = {
+        let cache = std::sync::Arc::clone(&cache);
+        tokio::spawn(async move {
+            let _ = cache.get::<User>("holder").await;
+        })
+    };
+    let deadline = std::time::Instant::now() + Duration::from_secs(5);
+    while handle.inner.calls.load(std::sync::atomic::Ordering::SeqCst) < 1 {
+        assert!(
+            std::time::Instant::now() < deadline,
+            "holder never reached the backend"
+        );
+        tokio::time::sleep(Duration::from_millis(5)).await;
+    }
+
+    // The macro's read is shed with Backpressure → the body must run
+    // uncached (the follow-up cache fill is shed too, and swallowed).
+    let user = shed_op(&cache, 7)
+        .await
+        .expect("a backpressure shed falls open like CircuitOpen");
+    assert_eq!(user.name, "shed 7");
+    assert_eq!(
+        SHED_RUNS.load(std::sync::atomic::Ordering::SeqCst),
+        1,
+        "the body ran uncached"
+    );
+    assert_eq!(
+        handle.inner.calls.load(std::sync::atomic::Ordering::SeqCst),
+        1,
+        "shed calls never reached the backend"
+    );
+
+    holder.abort();
+}
diff --git a/crates/cachekit/tests/reliability_tests.rs b/crates/cachekit/tests/reliability_tests.rs
index a630b10..ab5107c 100644
--- a/crates/cachekit/tests/reliability_tests.rs
+++ b/crates/cachekit/tests/reliability_tests.rs
@@ -19,7 +19,9 @@ use crate::common::MockBackend;
 use cachekit::backend::{Backend, HealthStatus, LockableBackend};
 use cachekit::client::SharedBackend;
 use cachekit::error::{BackendError, BackendErrorKind};
-use cachekit::reliability::{CircuitBreakerConfig, ReliabilityConfig, RetryConfig};
+use cachekit::reliability::{
+    BackpressureConfig, CircuitBreakerConfig, ReliabilityConfig, RetryConfig,
+};
 use cachekit::{CacheKit, CachekitError};
 
 // ── ScriptedBackend ──────────────────────────────────────────────────────────
@@ -154,6 +156,7 @@ async fn retry_recovers_from_transient_failures() {
         ReliabilityConfig {
             retry: Some(fast_retry(3)),
             circuit_breaker: None,
+            backpressure: None,
         },
     );
 
@@ -174,6 +177,7 @@ async fn retry_recovers_from_timeouts() {
         ReliabilityConfig {
             retry: Some(fast_retry(3)),
             circuit_breaker: None,
+            backpressure: None,
         },
     );
 
@@ -190,6 +194,7 @@ async fn retry_does_not_touch_permanent_errors() {
         ReliabilityConfig {
             retry: Some(fast_retry(3)),
             circuit_breaker: None,
+            backpressure: None,
         },
     );
 
@@ -209,6 +214,7 @@ async fn retry_exhausts_attempts_then_propagates() {
         ReliabilityConfig {
             retry: Some(fast_retry(3)),
             circuit_breaker: None,
+            backpressure: None,
         },
     );
 
@@ -231,6 +237,7 @@ async fn breaker_opens_and_fails_fast_without_touching_backend() {
                 open_timeout: Duration::from_secs(60),
                 ..CircuitBreakerConfig::default()
             }),
+            backpressure: None,
         },
     );
 
@@ -261,6 +268,7 @@ async fn breaker_half_open_probe_recovers() {
                 half_open_max_calls: 3,
                 rolling_window: Duration::from_secs(60),
             }),
+            backpressure: None,
         },
     );
 
@@ -295,6 +303,7 @@ async fn breaker_counts_one_failure_per_exhausted_retry_sequence() {
                 open_timeout: Duration::from_secs(60),
                 ..CircuitBreakerConfig::default()
             }),
+            backpressure: None,
         },
     );
 
@@ -585,6 +594,7 @@ async fn cancelled_probe_does_not_wedge_the_breaker() {
                 half_open_max_calls: 1,
                 rolling_window: Duration::from_secs(60),
             }),
+            backpressure: None,
         },
     );
 
@@ -610,3 +620,386 @@ async fn cancelled_probe_does_not_wedge_the_breaker() {
         "breaker recovered instead of wedging half-open"
     );
 }
+
+// ── Backpressure (LAB-729) ───────────────────────────────────────────────────
+
+fn bp(max_concurrent: usize, max_queue: usize, acquire_timeout: Duration) -> BackpressureConfig {
+    BackpressureConfig {
+        max_concurrent,
+        max_queue,
+        acquire_timeout,
+    }
+}
+
+/// Backend that records how many `get`s are in flight at once.
+#[derive(Debug, Default)]
+struct ProbeInner {
+    current: AtomicU32,
+    max_seen: AtomicU32,
+    calls: AtomicU32,
+}
+
+#[derive(Debug, Clone, Default)]
+struct ConcurrencyProbeBackend {
+    inner: Arc<ProbeInner>,
+}
+
+impl ConcurrencyProbeBackend {
+    fn new_with_handle() -> (SharedBackend, Self) {
+        let backend = Self::default();
+        let handle = backend.clone();
+        #[cfg(not(feature = "unsync"))]
+        let shared: SharedBackend = Arc::new(backend);
+        #[cfg(feature = "unsync")]
+        let shared: SharedBackend = std::rc::Rc::new(backend);
+        (shared, handle)
+    }
+}
+
+#[cfg_attr(not(feature = "unsync"), async_trait)]
+#[cfg_attr(feature = "unsync", async_trait(?Send))]
+impl Backend for ConcurrencyProbeBackend {
+    async fn get(&self, _key: &str) -> Result<Option<Vec<u8>>, BackendError> {
+        let now = self.inner.current.fetch_add(1, Ordering::SeqCst) + 1;
+        self.inner.max_seen.fetch_max(now, Ordering::SeqCst);
+        self.inner.calls.fetch_add(1, Ordering::SeqCst);
+        tokio::time::sleep(Duration::from_millis(20)).await;
+        self.inner.current.fetch_sub(1, Ordering::SeqCst);
+        Ok(Some(rmp_encoded_seven()))
+    }
+
+    async fn set(
+        &self,
+        _key: &str,
+        _value: Vec<u8>,
+        _ttl: Option<Duration>,
+    ) -> Result<(), BackendError> {
+        Ok(())
+    }
+
+    async fn delete(&self, _key: &str) -> Result<bool, BackendError> {
+        Ok(false)
+    }
+
+    async fn exists(&self, _key: &str) -> Result<bool, BackendError> {
+        Ok(false)
+    }
+
+    async fn health(&self) -> Result<HealthStatus, BackendError> {
+        Ok(HealthStatus {
+            is_healthy: true,
+            latency_ms: 0.0,
+            backend_type: "concurrency-probe".to_owned(),
+            details: HashMap::new(),
+        })
+    }
+}
+
+/// Backend whose `get` blocks until the test releases it (one release = one
+/// completed call). Counts calls that actually reach the backend.
+#[derive(Debug)]
+struct GateInner {
+    gate: tokio::sync::Semaphore,
+    calls: AtomicU32,
+}
+
+#[derive(Debug, Clone)]
+struct GateBackend {
+    inner: Arc<GateInner>,
+}
+
+impl GateBackend {
+    fn new_with_handle() -> (SharedBackend, Self) {
+        let backend = Self {
+            inner: Arc::new(GateInner {
+                gate: tokio::sync::Semaphore::new(0),
+                calls: AtomicU32::new(0),
+            }),
+        };
+        let handle = backend.clone();
+        #[cfg(not(feature = "unsync"))]
+        let shared: SharedBackend = Arc::new(backend);
+        #[cfg(feature = "unsync")]
+        let shared: SharedBackend = std::rc::Rc::new(backend);
+        (shared, handle)
+    }
+
+    fn release(&self, n: usize) {
+        self.inner.gate.add_permits(n);
+    }
+
+    fn calls(&self) -> u32 {
+        self.inner.calls.load(Ordering::SeqCst)
+    }
+}
+
+#[cfg_attr(not(feature = "unsync"), async_trait)]
+#[cfg_attr(feature = "unsync", async_trait(?Send))]
+impl Backend for GateBackend {
+    async fn get(&self, _key: &str) -> Result<Option<Vec<u8>>, BackendError> {
+        self.inner.calls.fetch_add(1, Ordering::SeqCst);
+        match self.inner.gate.acquire().await {
+            Ok(permit) => {
+                permit.forget();
+                Ok(Some(rmp_encoded_seven()))
+            }
+            Err(_) => Err(BackendError::permanent("gate closed")),
+        }
+    }
+
+    async fn set(
+        &self,
+        _key: &str,
+        _value: Vec<u8>,
+        _ttl: Option<Duration>,
+    ) -> Result<(), BackendError> {
+        Ok(())
+    }
+
+    async fn delete(&self, _key: &str) -> Result<bool, BackendError> {
+        Ok(false)
+    }
+
+    async fn exists(&self, _key: &str) -> Result<bool, BackendError> {
+        Ok(false)
+    }
+
+    async fn health(&self) -> Result<HealthStatus, BackendError> {
+        Ok(HealthStatus {
+            is_healthy: true,
+            latency_ms: 0.0,
+            backend_type: "gate".to_owned(),
+            details: HashMap::new(),
+        })
+    }
+}
+
+/// Poll until the gated backend has admitted at least one call. Bounded and
+/// relational (`< 1`, deadline) so a regression fails the test fast instead
+/// of hanging CI on an equality-terminated spin.
+async fn wait_for_backend_entry(handle: &GateBackend) {
+    let deadline = std::time::Instant::now() + Duration::from_secs(5);
+    while handle.calls() < 1 {
+        assert!(
+            std::time::Instant::now() < deadline,
+            "holder never reached the backend"
+        );
+        tokio::time::sleep(Duration::from_millis(5)).await;
+    }
+}
+
+#[tokio::test]
+async fn backpressure_caps_concurrent_backend_ops() {
+    // AC (LAB-729): with the cap at K, no more than K backend ops are ever in
+    // flight under a burst of ≫K concurrent callers — and nobody is shed as
+    // long as the waiting queue and timeout absorb the burst.
+    let (shared, handle) = ConcurrencyProbeBackend::new_with_handle();
+    let client = Arc::new(client_with(
+        shared,
+        ReliabilityConfig {
+            retry: None,
+            circuit_breaker: None,
+            backpressure: Some(bp(4, 1000, Duration::from_secs(5))),
+        },
+    ));
+
+    let tasks: Vec<_> = (0..32)
+        .map(|_| {
+            let client = Arc::clone(&client);
+            tokio::spawn(async move { client.get::<u32>("k").await })
+        })
+        .collect();
+    for task in tasks {
+        let value = task.await.expect("task").expect("get succeeds");
+        assert_eq!(value, Some(7));
+    }
+
+    assert_eq!(handle.inner.calls.load(Ordering::SeqCst), 32);
+    let max_seen = handle.inner.max_seen.load(Ordering::SeqCst);
+    assert!(max_seen <= 4, "cap 4 exceeded: {max_seen} ops in flight");
+}
+
+#[tokio::test]
+async fn backpressure_sheds_immediately_when_queue_disabled() {
+    // max_queue: 0 → a saturated limiter sheds on the spot with the typed
+    // Backpressure error, without waiting out acquire_timeout and without
+    // the call ever reaching the backend. Also the regression test for the
+    // builder gating: a backpressure-ONLY config must still wrap the backend.
+    let (shared, handle) = GateBackend::new_with_handle();
+    let client = Arc::new(client_with(
+        shared,
+        ReliabilityConfig {
+            retry: None,
+            circuit_breaker: None,
+            backpressure: Some(bp(1, 0, Duration::from_secs(5))),
+        },
+    ));
+
+    let holder = {
+        let client = Arc::clone(&client);
+        tokio::spawn(async move { client.get::<u32>("k").await })
+    };
+    // Let the holder reach the backend and park on the gate.
+    wait_for_backend_entry(&handle).await;
+
+    let start = std::time::Instant::now();
+    let err = client
+        .get::<u32>("k")
+        .await
+        .expect_err("saturated limiter with no queue sheds");
+    assert_eq!(backend_kind(&err), Some(&BackendErrorKind::Backpressure));
+    assert!(
+        start.elapsed() < Duration::from_millis(500),
+        "queue-full sheds immediately, not after the 5 s acquire_timeout"
+    );
+    assert_eq!(handle.calls(), 1, "the shed call never reached the backend");
+
+    handle.release(1);
+    let value = holder.await.expect("task").expect("holder completes");
+    assert_eq!(value, Some(7));
+}
+
+#[tokio::test]
+async fn backpressure_times_out_waiting_then_sheds() {
+    // A queued caller waits acquire_timeout for a permit, then is shed with
+    // the typed Backpressure error (bounded wait, never an unbounded queue).
+    let (shared, handle) = GateBackend::new_with_handle();
+    let client = Arc::new(client_with(
+        shared,
+        ReliabilityConfig {
+            retry: None,
+            circuit_breaker: None,
+            backpressure: Some(bp(1, 10, Duration::from_millis(50))),
+        },
+    ));
+
+    let holder = {
+        let client = Arc::clone(&client);
+        tokio::spawn(async move { client.get::<u32>("k").await })
+    };
+    wait_for_backend_entry(&handle).await;
+
+    let start = std::time::Instant::now();
+    let err = client
+        .get::<u32>("k")
+        .await
+        .expect_err("no permit frees up within acquire_timeout");
+    assert_eq!(backend_kind(&err), Some(&BackendErrorKind::Backpressure));
+    assert!(
+        start.elapsed() >= Duration::from_millis(40),
+        "the caller joins the queue and waits out acquire_timeout"
+    );
+    assert_eq!(handle.calls(), 1, "the shed call never reached the backend");
+
+    handle.release(1);
+    holder.await.expect("task").expect("holder completes");
+}
+
+#[tokio::test]
+async fn backpressure_rejections_do_not_open_the_breaker() {
+    // Shed calls are caller-side overload, not backend-health signals: with
+    // failure_threshold: 1, a single counted failure would open the circuit —
+    // so the success after the sheds proves they never touched the breaker.
+    let (shared, handle) = GateBackend::new_with_handle();
+    let client = Arc::new(client_with(
+        shared,
+        ReliabilityConfig {
+            retry: None,
+            circuit_breaker: Some(CircuitBreakerConfig {
+                failure_threshold: 1,
+                open_timeout: Duration::from_secs(60),
+                ..CircuitBreakerConfig::default()
+            }),
+            backpressure: Some(bp(1, 0, Duration::from_secs(5))),
+        },
+    ));
+
+    let holder = {
+        let client = Arc::clone(&client);
+        tokio::spawn(async move { client.get::<u32>("k").await })
+    };
+    wait_for_backend_entry(&handle).await;
+
+    for _ in 0..3 {
+        let err = client
+            .get::<u32>("k")
+            .await
+            .expect_err("shed while saturated");
+        assert_eq!(backend_kind(&err), Some(&BackendErrorKind::Backpressure));
+    }
+
+    handle.release(1);
+    holder.await.expect("task").expect("holder completes");
+
+    // Were sheds counted as breaker failures, this would be CircuitOpen.
+    handle.release(1);
+    let value = client
+        .get::<u32>("k")
+        .await
+        .expect("breaker stayed closed through the sheds");
+    assert_eq!(value, Some(7));
+    assert_eq!(handle.calls(), 2, "sheds never reached the backend");
+}
+
+#[tokio::test]
+async fn backpressure_composes_with_retry_without_deadlock() {
+    // The permit is held across the whole retry sequence (backoff included).
+    // With the cap at 1 and two concurrent callers, the second caller queues
+    // behind the first's full retry sequence and both complete — permits
+    // held across backoff cannot deadlock because a permit holder never
+    // re-enters the limiter.
+    let (shared, handle) = ScriptedBackend::new_with_handle(2, BackendErrorKind::Transient);
+    let client = Arc::new(client_with(
+        shared,
+        ReliabilityConfig {
+            retry: Some(fast_retry(3)),
+            circuit_breaker: None,
+            backpressure: Some(bp(1, 10, Duration::from_secs(5))),
+        },
+    ));
+
+    let a = {
+        let client = Arc::clone(&client);
+        tokio::spawn(async move { client.get::<u32>("k").await })
+    };
+    let b = {
+        let client = Arc::clone(&client);
+        tokio::spawn(async move { client.get::<u32>("k").await })
+    };
+
+    let a = a.await.expect("task a").expect("a succeeds");
+    let b = b.await.expect("task b").expect("b succeeds");
+    assert_eq!(a, Some(7));
+    assert_eq!(b, Some(7));
+    assert_eq!(
+        handle.calls(),
+        4,
+        "one caller retries through 2 failures (3 attempts), the other hits once"
+    );
+}
+
+#[tokio::test]
+async fn backpressure_permit_released_after_error() {
+    // A failed operation must release its permit: with the cap at 1, the
+    // second sequential call would otherwise be shed instead of reaching the
+    // backend for its own (transient) failure.
+    let (shared, handle) = ScriptedBackend::new_with_handle(u32::MAX, BackendErrorKind::Transient);
+    let client = client_with(
+        shared,
+        ReliabilityConfig {
+            retry: None,
+            circuit_breaker: None,
+            backpressure: Some(bp(1, 0, Duration::from_millis(50))),
+        },
+    );
+
+    for _ in 0..2 {
+        let err = client.get::<u32>("k").await.expect_err("backend failing");
+        assert_eq!(
+            backend_kind(&err),
+            Some(&BackendErrorKind::Transient),
+            "a Backpressure kind here would mean the first call leaked its permit"
+        );
+    }
+    assert_eq!(handle.calls(), 2);
+}