tensor_wasm_tenant/context.rs
1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 Craton Software Company
3//! `TenantContext`: per-tenant CUDA context + stream + memory pool.
4//!
5//! A `TenantContext` is the runtime handle that ties a [`TenantId`] to the GPU
6//! resources reserved for that tenant: a CUDA stream identifier, an isolation
7//! policy ([`IsolationKind`]), and a memory quota that the scheduler enforces
8//! before kernels are dispatched. Construction goes through
9//! [`TenantContextBuilder`] so callers can opt into individual fields without
10//! a 5-argument constructor.
11//!
12//! Under the `cuda` feature, each `ContextIsolated` tenant additionally owns a
13//! real `cust::context::Context`; without that feature (the default on
14//! CUDA-less hosts), the field collapses to a unit stub so the rest of the
15//! crate compiles and tests run unchanged.
16//!
17//! NOTE: cuda-feature code in this file is compile-tested on CUDA hosts only;
18//! on no-CUDA hosts only the `#[cfg(not(feature = "cuda"))]` branches are
19//! exercised. The cuda branches use the `cust` 0.3.x context-stack and
20//! primary-context APIs.
21
22// The `loom` feature swaps `std::sync::atomic::AtomicU64` for
23// `loom::sync::atomic::AtomicU64` so `tests/loom_consume_release.rs` can
24// drive `consume_bytes_inner` / `release_bytes_inner` through loom's
25// exhaustive scheduler. The two types share the same surface API used
26// by the CAS loops below (`load`, `compare_exchange_weak`, `fetch_add`,
27// `new`), so the inner-loop bodies need no further cfg-gating.
28//
29// NOTE(loom): `loom::sync::atomic::AtomicU64::new` is NOT `const fn`
30// (loom's atomics carry per-execution tracking state), whereas
31// `std::sync::atomic::AtomicU64::new` is. The `static
32// ISOLATION_DOWNGRADE_COUNT` declaration below therefore needs the
33// std-flavoured type even under `--features loom`; the eventual
34// loom model body in `tests/loom_consume_release.rs` builds a
35// dedicated `loom::sync::atomic::AtomicU64` (or a minimal stand-in
36// struct that uses one) rather than reaching for this static. This
37// scaffold keeps the import swap localised to the CAS-loop hot path
38// while leaving the alert-counter on the std type for static-init
39// compatibility.
40#[cfg(feature = "loom")]
41use loom::sync::atomic::{AtomicU64, Ordering};
42#[cfg(not(feature = "loom"))]
43use std::sync::atomic::{AtomicU64, Ordering};
44
45use std::sync::{Arc, Mutex};
46use std::time::{Duration, Instant};
47
48use tensor_wasm_core::error::TensorWasmError;
49use tensor_wasm_core::mem_pool::DriverMemPool;
50use tensor_wasm_core::metrics::{TenantLabels, TensorWasmMetrics};
51use tensor_wasm_core::types::TenantId;
52
53/// Process-wide count of `IsolationKind::ContextIsolated` requests that
54/// could not be honoured by the CUDA driver and were silently downgraded
55/// to `IsolationKind::StreamIsolated` at [`TenantContextBuilder::build`]
56/// time. Operators that requested context isolation as a deployment
57/// constraint (e.g. multi-tenant untrusted workloads on a shared GPU)
58/// should alert on any non-zero reading — the downgrade is honest
59/// reporting at the type level, but it is also a deployment-config bug
60/// that needs to be surfaced. Incremented at most once per failed
61/// build; never decremented.
62///
63/// Read via [`isolation_downgrade_count`]. Not yet exported through the
64/// `prometheus-client` registry in `tensor-wasm-core`: as of this crate
65/// version [`tensor_wasm_core::metrics::TensorWasmMetrics`] exposes no
66/// counter whose semantics match a per-process isolation-downgrade tally
67/// (the existing `Counter<u64>` accessors — `kernel_dispatches_total`,
68/// `offload_fallback_total`, etc. — all carry unrelated meaning, and
69/// reusing one would corrupt those series), and that crate is owned by a
70/// separate component that this crate must not edit. The metric is
71/// therefore intentionally cheap (a single `AtomicU64`) and lives at the
72/// call site, surfaced via [`isolation_downgrade_count`] so the alert
73/// pipeline can scrape it out-of-band today.
74///
75/// TODO(core): wire registry export once `tensor-wasm-core` grows a
76/// dedicated counter. The minimal core-crate API required is a
77/// `Counter<u64>` field on `TensorWasmMetrics` registered as
78/// `tensor_wasm_isolation_downgrade_total` with a public accessor
79/// `TensorWasmMetrics::isolation_downgrade_total(&self) -> &Counter<u64>`.
80/// When that lands, the `build()` downgrade path below should call
81/// `metrics.isolation_downgrade_total().inc()` whenever `self.metrics`
82/// is `Some`, in addition to bumping this static.
83#[cfg(not(feature = "loom"))]
84static ISOLATION_DOWNGRADE_COUNT: AtomicU64 = AtomicU64::new(0);
85
86/// Process-wide count of `ContextIsolated -> StreamIsolated` downgrades
87/// observed since startup. See `ISOLATION_DOWNGRADE_COUNT` for the
88/// alert contract: any non-zero reading on an operator that requested
89/// `ContextIsolated` is a deployment-config bug.
90///
91/// Under the `loom` feature `loom::sync::atomic::AtomicU64::new` is not
92/// `const fn`, so the static counter is suppressed and this returns 0.
93/// The loom test harness builds its own atomic per-execution.
94#[cfg(not(feature = "loom"))]
95pub fn isolation_downgrade_count() -> u64 {
96 ISOLATION_DOWNGRADE_COUNT.load(Ordering::Relaxed)
97}
98/// Loom-build stub for [`isolation_downgrade_count`]; always returns 0.
99#[cfg(feature = "loom")]
100pub fn isolation_downgrade_count() -> u64 {
101 0
102}
103
104/// How aggressively a tenant's GPU work is separated from other tenants'.
105///
106/// The variants mirror the levels exposed by `tensor-wasm-mem::isolation::IsolationLevel`,
107/// but live here as a separate type so this crate can be consumed without
108/// pulling in the Wasmtime-dependent memory crate.
109#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Hash)]
110pub enum IsolationKind {
111 /// All tenants share the default CUDA context and stream.
112 ///
113 /// Cheap to spawn but unsuitable for multi-tenant untrusted workloads.
114 Shared,
115 /// Each tenant gets its own CUDA stream; contexts are shared.
116 ///
117 /// Default for multi-tenant deployments — prevents kernel-ordering
118 /// accidents without paying the cost of per-tenant context creation.
119 #[default]
120 StreamIsolated,
121 /// Each tenant gets its own CUDA context (via MPS when available, or
122 /// `cuCtxCreate` otherwise).
123 ContextIsolated,
124}
125
126impl IsolationKind {
127 /// Stable, human-readable name (used in span attributes and metrics).
128 pub fn name(self) -> &'static str {
129 match self {
130 IsolationKind::Shared => "shared",
131 IsolationKind::StreamIsolated => "stream_isolated",
132 IsolationKind::ContextIsolated => "context_isolated",
133 }
134 }
135}
136
137impl std::fmt::Display for IsolationKind {
138 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
139 f.write_str(self.name())
140 }
141}
142
143/// Returned by [`TenantContext::try_acquire_op`] /
144/// [`TenantContext::try_acquire_ops`] when a tenant has exhausted its
145/// time-windowed operation-rate budget.
146///
147/// Distinct from the byte-cap errors (`TensorWasmError::MemoryExhausted` /
148/// `GpuMemoryExhausted`), which are high-water-mark *capacity* refusals: a
149/// rate-limit refusal is transient — the same request will succeed once the
150/// token bucket refills. The struct carries enough context for a scheduler to
151/// back off intelligently (how many tokens were asked for, how many were
152/// available, and the configured steady-state rate) and is deliberately a
153/// crate-local type rather than a new `TensorWasmError` variant: the rate
154/// limiter is an additive, opt-in noisy-neighbour control that this crate owns
155/// end-to-end, and `tensor-wasm-core` (which owns `TensorWasmError`) is a
156/// separate component this crate must not edit.
157#[derive(Debug, Clone, Copy, PartialEq, Eq)]
158pub struct RateLimited {
159 /// Tokens (operations) the caller requested.
160 pub requested: u64,
161 /// Tokens available in the bucket at the moment of refusal (after the
162 /// time-based refill was applied). Always `< requested`.
163 pub available: u64,
164 /// Configured steady-state refill rate in tokens per second
165 /// (`ops_per_sec` passed to [`TenantContextBuilder::with_rate_limit`]).
166 pub ops_per_sec: u64,
167 /// Configured bucket depth (`burst` passed to
168 /// [`TenantContextBuilder::with_rate_limit`]).
169 pub burst: u64,
170}
171
172impl std::fmt::Display for RateLimited {
173 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
174 write!(
175 f,
176 "rate limited: requested {} ops, {} available (rate {}/s, burst {})",
177 self.requested, self.available, self.ops_per_sec, self.burst
178 )
179 }
180}
181
182impl std::error::Error for RateLimited {}
183
184/// A monotonic-clock token bucket guarding a tenant's operation rate.
185///
186/// The bucket holds up to `burst` tokens and refills at `ops_per_sec` tokens
187/// per second, computed lazily from the elapsed time since the last
188/// observation — there is no background timer thread. Each admitted operation
189/// removes one (or `n`) tokens; a request that cannot be fully satisfied is
190/// rejected with [`RateLimited`] and removes *no* tokens (all-or-nothing), so
191/// a burst of large requests cannot partially drain the bucket and starve
192/// smaller ones mid-flight.
193///
194/// State is `Mutex`-guarded rather than a lock-free atomic pair: the refill
195/// math reads the last-refill instant and the token count together and writes
196/// both back, which is a small critical section but genuinely needs to be
197/// atomic *as a unit*. The lock is held only for the duration of that
198/// arithmetic (no I/O, no syscalls beyond `Instant::now`), so contention is
199/// negligible next to the work an admitted operation actually goes on to do.
200/// Scale factor for the fixed-point token accounting: tokens are tracked
201/// internally in *micro-tokens* (1 whole token == `MICRO_PER_TOKEN`
202/// micro-tokens). All refill / consume / cap arithmetic is therefore exact
203/// integer arithmetic on `u64`, with no `f64` rounding or epsilon fudge.
204///
205/// At micro-token granularity, sub-token refill credit (a 100 ops/s bucket
206/// accruing 1 token every 10 ms) is preserved between calls exactly as the
207/// old `f64` `tokens` field intended, but an exactly-full bucket compares
208/// *exactly* equal to its capacity — so a full bucket always admits, and
209/// accumulated rounding can no longer spuriously reject it.
210///
211/// `1_000_000` gives microsecond-equivalent resolution: even a 1 ops/s rate
212/// credits 1 micro-token per microsecond of elapsed time, far finer than any
213/// realistic scheduler cares about, while keeping the `u64` headroom huge
214/// (`u64::MAX / 1_000_000` ≈ 1.8e13 whole tokens of burst capacity).
215const MICRO_PER_TOKEN: u64 = 1_000_000;
216
217#[derive(Debug)]
218struct TokenBucket {
219 /// Steady-state refill rate, tokens per second. Refill credit is computed
220 /// as exact integer micro-tokens from the elapsed `Duration` (see
221 /// [`TokenBucket::refilled_micros`]), so sub-token credit is carried on
222 /// `TokenBucketState::micro_tokens` without any floating-point rounding.
223 ops_per_sec: u64,
224 /// Maximum *whole* tokens the bucket can hold (the burst depth). The
225 /// public [`RateLimited::burst`] reports this value unchanged.
226 burst: u64,
227 /// Bucket capacity in micro-tokens (`burst * MICRO_PER_TOKEN`, saturating
228 /// so an absurdly large burst cannot overflow `u64`). Precomputed once so
229 /// the hot path is a plain `min`.
230 capacity_micros: u64,
231 /// Mutable bucket state, guarded as a unit.
232 state: Mutex<TokenBucketState>,
233}
234
235#[derive(Debug)]
236struct TokenBucketState {
237 /// Current token count in micro-tokens, preserving fractional refill
238 /// credit between observations as an exact integer (no `f64` rounding).
239 micro_tokens: u64,
240 /// Last instant the bucket was refilled. Advanced on every
241 /// `try_acquire`, so the next call only credits time since this point.
242 last_refill: Instant,
243}
244
245impl TokenBucket {
246 /// Construct a bucket that starts full (`burst` tokens available), so the
247 /// first `burst` operations are admitted immediately before any refill is
248 /// needed. `ops_per_sec` and `burst` are both clamped to a minimum of 1
249 /// so a `with_rate_limit(0, 0)` cannot wedge the tenant into a
250 /// never-admits state — the builder documents that "no rate limit" is
251 /// expressed by *not* calling `with_rate_limit` at all (the `None` default).
252 fn new(ops_per_sec: u64, burst: u64) -> Self {
253 let burst = burst.max(1);
254 let ops_per_sec = ops_per_sec.max(1);
255 // Saturating so a pathologically large burst (> u64::MAX / MICRO)
256 // cannot overflow the micro-token capacity. Realistic bursts are far
257 // below this ceiling, so the saturation is a safety net, not a
258 // behavioural change.
259 let capacity_micros = burst.saturating_mul(MICRO_PER_TOKEN);
260 Self {
261 ops_per_sec,
262 burst,
263 capacity_micros,
264 state: Mutex::new(TokenBucketState {
265 // Starts full: `burst` whole tokens == `capacity_micros`.
266 micro_tokens: capacity_micros,
267 last_refill: Instant::now(),
268 }),
269 }
270 }
271
272 /// Micro-token balance after crediting `elapsed` of refill onto `current`,
273 /// capped at the bucket capacity.
274 ///
275 /// Refill credit in micro-tokens is
276 /// `elapsed_nanos * ops_per_sec * MICRO_PER_TOKEN / 1_000_000_000`. The
277 /// intermediate product is computed in `u128` so even a multi-year
278 /// `elapsed` cannot overflow before the per-nanosecond divide; the result
279 /// is then saturated back into `u64` and capped at `capacity_micros`, so
280 /// a long idle period simply tops the bucket off at its burst depth (and
281 /// never wraps). The division truncates toward zero, which only ever
282 /// *under*-credits by at most one micro-token — it can never manufacture
283 /// tokens that would let an over-budget request slip through.
284 fn refilled_micros(&self, current: u64, elapsed: Duration) -> u64 {
285 let nanos = elapsed.as_nanos();
286 // credit_micros = nanos * ops_per_sec * MICRO_PER_TOKEN / 1e9
287 let credit_micros = nanos
288 .saturating_mul(self.ops_per_sec as u128)
289 .saturating_mul(MICRO_PER_TOKEN as u128)
290 / 1_000_000_000u128;
291 let credit_micros = u64::try_from(credit_micros).unwrap_or(u64::MAX);
292 current
293 .saturating_add(credit_micros)
294 .min(self.capacity_micros)
295 }
296
297 /// Attempt to remove `n` tokens, refilling for elapsed wall-clock time
298 /// first. Admits (returns `Ok`) only if at least `n` tokens are available
299 /// after the refill; otherwise returns [`RateLimited`] and leaves the
300 /// bucket untouched (all-or-nothing).
301 ///
302 /// `now` is injected rather than read internally so tests can drive the
303 /// refill deterministically without sleeping; the public
304 /// [`TenantContext::try_acquire_op`] passes `Instant::now()`.
305 ///
306 /// All accounting is exact integer arithmetic in micro-tokens, so the
307 /// admission test is a plain `available_micros >= need_micros`: an
308 /// exactly-full bucket has `available_micros == capacity_micros` and a
309 /// request for exactly the full burst has `need_micros == capacity_micros`,
310 /// which compares equal and therefore admits (no `f64::EPSILON` nudge, no
311 /// magnitude-dependent rounding that could spuriously reject it).
312 fn try_acquire_at(&self, n: u64, now: Instant) -> Result<(), RateLimited> {
313 let mut state = self
314 .state
315 .lock()
316 .unwrap_or_else(|poisoned| poisoned.into_inner());
317 // `now` may predate `last_refill` only if a caller injected a
318 // non-monotonic instant in a test; `checked_duration_since` yields
319 // `None` there and we credit nothing rather than panicking.
320 let elapsed = now
321 .checked_duration_since(state.last_refill)
322 .unwrap_or(Duration::ZERO);
323 let available_micros = self.refilled_micros(state.micro_tokens, elapsed);
324 // `need` in micro-tokens. Saturating so a colossal `n` cannot overflow;
325 // such a request necessarily exceeds the bucket and is rejected below.
326 let need_micros = n.saturating_mul(MICRO_PER_TOKEN);
327 if available_micros < need_micros {
328 return Err(RateLimited {
329 requested: n,
330 // Report the whole tokens that are actually acquirable right
331 // now (floor of the micro-token balance). This is consistent
332 // with the admission decision: the request was rejected
333 // precisely because `available_micros < need_micros`, so the
334 // floored whole-token count is strictly less than `n` and the
335 // documented `available < requested` invariant holds. Floor —
336 // not the old `f64 as u64` truncation of a possibly-rounded
337 // value — is exact, so we never under-report a token the
338 // caller could in fact have acquired.
339 available: available_micros / MICRO_PER_TOKEN,
340 ops_per_sec: self.ops_per_sec,
341 burst: self.burst,
342 });
343 }
344 state.micro_tokens = available_micros - need_micros;
345 state.last_refill = now;
346 Ok(())
347 }
348}
349
350/// Unforgeable proof of authority to mutate a single tenant's quota counters.
351///
352/// Minted only by [`crate::TenantRegistry::register_with_capability`]; the
353/// `_seal` field is private to this crate so no downstream crate (and no
354/// hostile workload) can construct one out of thin air. Holding an
355/// `Arc<TenantContext>` is therefore no longer sufficient to drive that
356/// tenant's `bytes_in_use` counter — the caller must also present the
357/// matching capability, which it can only get if it originally registered
358/// the tenant.
359///
360/// `Clone` is intentionally derived: the API gateway holds the
361/// authoritative copy and may need to hand clones to per-tenant subsystems
362/// (the scheduler, the memory pool, etc.). What is NOT derived is any
363/// `From<TenantId>` or public constructor, so a workload running inside
364/// tenant A cannot fabricate one for tenant B.
365///
366/// # Registry binding (H1 — always enforced)
367///
368/// Every capability carries an `Arc<()>` token that points to its
369/// minting registry's allocation. Comparison is by `Arc::ptr_eq`, so a
370/// capability minted by registry A is rejected when presented against a
371/// context registered in registry B, even if both contexts happen to
372/// share the same numeric `TenantId`. Without this binding, capabilities
373/// from independent registries are interchangeable, which the H1 audit
374/// finding flagged as a cross-registry capability-confusion vector — see
375/// the note on `RegistryAdminCapability` for the threat model.
376///
377/// H1 fix: this binding is now UNCONDITIONAL and no longer behind the
378/// `strict-cap-binding` feature gate. A release build with the feature
379/// disabled still binds caps to their minting registry and still fails
380/// closed on cross-registry / forged-admin caps. The `strict-cap-binding`
381/// feature remains defined only to keep the typed `*_strict` admin APIs
382/// and the `cap_binding_strict` integration test compiling.
383#[derive(Debug, Clone)]
384pub struct TenantCapability {
385 tenant_id: TenantId,
386 /// Crate-private zero-sized seal: prevents `TenantCapability { .. }`
387 /// struct-literal construction outside `tensor-wasm-tenant`.
388 _seal: (),
389 /// Pointer-identity stamp of the registry that minted this capability.
390 /// H1: always present so registry binding is enforced unconditionally.
391 pub(crate) registry_token: std::sync::Arc<()>,
392}
393
394impl TenantCapability {
395 /// Mint a capability bound to `tenant_id` AND to the minting registry,
396 /// identified by `registry_token` (an `Arc::clone` of the registry's
397 /// per-instance token allocation). Comparison at `check_capability`
398 /// time is by `Arc::ptr_eq`, so two registries that happen to allocate
399 /// `Arc::new(())` at the same address would still be distinct
400 /// allocations and `ptr_eq` would return `false` for caps from one
401 /// against contexts of the other.
402 ///
403 /// H1: the registry-bound signature is now unconditional — there is no
404 /// longer a token-less `mint` variant, so a cap can never be minted
405 /// without provenance regardless of the `strict-cap-binding` feature.
406 pub(crate) fn mint(tenant_id: TenantId, registry_token: std::sync::Arc<()>) -> Self {
407 Self {
408 tenant_id,
409 _seal: (),
410 registry_token,
411 }
412 }
413
414 /// Identifier of the tenant this capability authorises.
415 pub fn tenant_id(&self) -> TenantId {
416 self.tenant_id
417 }
418}
419
420/// Per-tenant runtime handle: identity, isolation level, stream, and quota.
421///
422/// Instances are constructed through [`TenantContextBuilder`] and then placed
423/// into the [`crate::TenantRegistry`]. The byte-counter methods
424/// ([`Self::consume_bytes`] / [`Self::release_bytes`]) are lock-free and safe
425/// to call from any thread; quota enforcement happens at the point of
426/// allocation, not asynchronously.
427#[derive(Debug)]
428pub struct TenantContext {
429 tenant_id: TenantId,
430 isolation: IsolationKind,
431 stream_id: u64,
432 memory_quota_bytes: u64,
433 bytes_in_use: AtomicU64,
434
435 /// Maximum GPU memory in bytes this tenant may allocate concurrently.
436 /// `None` = no GPU memory cap (operator trust).
437 ///
438 /// ADVISORY / IN-PROCESS-ONLY: unless a [`DriverMemPool`] is wired in
439 /// via [`TenantContextBuilder::with_driver_enforced_gpu_cap`], this
440 /// cap is enforced ONLY by the in-process [`Self::gpu_bytes_in_use`]
441 /// counter on the [`Self::consume_gpu_bytes`] path. The CUDA driver
442 /// itself sees no cap, so any tenant that obtains a raw CUDA handle
443 /// and allocates outside `consume_gpu_bytes` is NOT capped. v0.3.7
444 /// records and reports usage; v0.4 enforces via cuMemPool's
445 /// `cuMemPoolSetAttribute(CU_MEMPOOL_ATTR_RELEASE_THRESHOLD, ...)`.
446 ///
447 /// Distinct from [`Self::memory_quota_bytes`]: that one is the
448 /// host-side / CPU quota the executor enforces against
449 /// `consume_bytes`. This field is consulted by the GPU allocator
450 /// path (`tensor-wasm-mem::TensorWasmMemoryCreator::with_tenant_context`)
451 /// against `gpu_bytes_in_use` on every `UnifiedBuffer::new_on`.
452 gpu_memory_bytes_cap: Option<u64>,
453 /// Bytes currently accounted as in-use against the GPU cap. Mirrors
454 /// the CPU [`Self::bytes_in_use`] counter; updated by
455 /// [`Self::consume_gpu_bytes`] / [`Self::release_gpu_bytes`] via the
456 /// same CAS-loop pattern, and (when a metrics handle is wired in
457 /// via [`TenantContextBuilder::with_metrics`]) republished as the
458 /// per-tenant series of
459 /// [`tensor_wasm_core::metrics::TensorWasmMetrics::gpu_memory_bytes_per_tenant`]
460 /// on every transition. v0.3.7 record-only: the value is the source
461 /// of truth for the in-process refusal of over-cap allocations. The
462 /// `gpu-mem-pool` `TenantMemPool` additionally enforces the cap
463 /// host-side in its `allocate` method; note that the CUDA driver itself
464 /// has no per-pool allocation ceiling
465 /// (`CU_MEMPOOL_ATTR_RELEASE_THRESHOLD` is a retention hint, not a cap —
466 /// see [`Self::gpu_memory_bytes_cap`]).
467 gpu_bytes_in_use: AtomicU64,
468
469 /// Recorded-only CUDA memory-pool release-threshold value. `None`
470 /// means "use the driver default" (typically unbounded retention).
471 ///
472 /// ADVISORY / RECORD-ONLY: this value is NEVER enforced. The cust
473 /// 0.3.x crate does not expose the `cuMemPool*` API, so this field is
474 /// **not** wired through to
475 /// `cudaMemPoolSetAttribute(CU_MEMPOOL_ATTR_RELEASE_THRESHOLD)` — it
476 /// is stored purely for inspection / metrics / forward-compat and the
477 /// CUDA driver never sees it. The in-process
478 /// [`TenantContext::bytes_in_use`] counter is the only enforcement of
479 /// this crate's quota. See
480 /// [`TenantContextBuilder::with_recorded_cuda_mem_pool_quota`] for
481 /// the honest naming and the upgrade path.
482 #[allow(dead_code)]
483 cuda_mem_pool_quota_bytes: Option<u64>,
484
485 /// Optional driver-level memory pool whose release threshold is pinned
486 /// to this tenant's GPU cap. When present, [`Self::consume_gpu_bytes`]
487 /// pushes the cap through
488 /// [`tensor_wasm_core::mem_pool::DriverMemPool::set_release_threshold`]
489 /// so the CUDA driver itself rejects over-cap allocations — closing
490 /// the bypass that the in-process `gpu_bytes_in_use` counter alone
491 /// cannot (a tenant who obtained a raw CUDA driver handle).
492 ///
493 /// Stored as a trait object (`Arc<dyn DriverMemPool>`), NOT the
494 /// concrete `tensor-wasm-mem::cuda_mem_pool::TenantMemPool`: this
495 /// crate depends only on `tensor-wasm-core`, never on
496 /// `tensor-wasm-mem`. Holding the concrete type would require a
497 /// `tenant -> mem` dependency edge, which would close the
498 /// `mem` <-> `tenant` cycle (mem already depends on tenant). The
499 /// backend-agnostic trait lives in `tensor-wasm-core`, which both
500 /// crates already depend on, so the graph stays acyclic. Set via
501 /// [`TenantContextBuilder::with_driver_enforced_gpu_cap`].
502 driver_mem_pool: Option<Arc<dyn DriverMemPool>>,
503
504 /// Optional time-windowed operation-rate limiter. `None` (the default)
505 /// preserves the historical pure high-water-mark byte-cap behaviour —
506 /// no per-operation rate is enforced. When set via
507 /// [`TenantContextBuilder::with_rate_limit`], every
508 /// [`Self::try_acquire_op`] / [`Self::try_acquire_ops`] consults a
509 /// monotonic-clock token bucket and refuses (with [`RateLimited`]) once
510 /// the tenant's steady-state rate plus burst is exceeded, addressing
511 /// noisy-neighbour scheduling without touching the byte counters above.
512 rate_limiter: Option<TokenBucket>,
513
514 // Real `cust::context::Context` under the `cuda` feature; otherwise a
515 // unit stub so the rest of the crate compiles on CUDA-less hosts.
516 #[cfg(feature = "cuda")]
517 #[allow(dead_code)]
518 cu_context: Option<cust::context::Context>,
519 #[cfg(not(feature = "cuda"))]
520 #[allow(dead_code)]
521 cu_context: (),
522
523 /// Optional shared metrics handle. When present, every CPU-side
524 /// [`Self::consume_bytes`] / [`Self::release_bytes`] transition updates
525 /// the per-tenant series of
526 /// [`tensor_wasm_core::metrics::TensorWasmMetrics::cpu_memory_bytes_per_tenant`],
527 /// and every GPU-side
528 /// [`Self::consume_gpu_bytes`] / [`Self::release_gpu_bytes`] transition
529 /// updates the per-tenant series of
530 /// [`tensor_wasm_core::metrics::TensorWasmMetrics::gpu_memory_bytes_per_tenant`].
531 /// The two no longer collide on a single labelled series. `None` keeps
532 /// the historical no-op behaviour so embedders that construct a
533 /// `TenantContext` outside the API gateway (e.g. benches, examples) do
534 /// not need to plumb a metrics registry.
535 metrics: Option<TensorWasmMetrics>,
536 /// Memoized label tuple used to address the per-tenant gauge series.
537 /// Built once at construction so the hot path of `consume_bytes` /
538 /// `release_bytes` does not allocate on every transition.
539 metrics_labels: TenantLabels,
540 /// Pointer-identity stamp of the registry this context was registered
541 /// in, used by [`Self::check_capability`] to reject caps minted by a
542 /// *different* registry.
543 ///
544 /// `None` until the context is moved into
545 /// [`crate::TenantRegistry::register_with_capability`], which sets
546 /// the token before wrapping the context in an `Arc`. A `None` here
547 /// at `check_capability` time means the context was never registered
548 /// (or was constructed for a test that bypasses the registry); the
549 /// registry-binding check then **fails closed** (rejects the cap),
550 /// since an unregistered context has no provenance to vouch for any
551 /// capability — see [`Self::check_capability`].
552 ///
553 /// H1: always present so registry binding is enforced unconditionally,
554 /// not just under the `strict-cap-binding` feature.
555 pub(crate) registry_token: Option<std::sync::Arc<()>>,
556}
557
558impl TenantContext {
559 /// Start a builder for a tenant with the given identifier.
560 pub fn builder(tenant_id: TenantId) -> TenantContextBuilder {
561 TenantContextBuilder::new(tenant_id)
562 }
563
564 /// Tenant identifier this context belongs to.
565 pub fn id(&self) -> TenantId {
566 self.tenant_id
567 }
568
569 /// Isolation level configured for this tenant.
570 pub fn isolation(&self) -> IsolationKind {
571 self.isolation
572 }
573
574 /// Stream identifier (logical handle; the actual `CUstream` lives in
575 /// `tensor-wasm-mem` / `tensor-wasm-wasi-gpu` and is keyed by this value).
576 pub fn stream_id(&self) -> u64 {
577 self.stream_id
578 }
579
580 /// Total bytes the tenant is permitted to allocate concurrently.
581 pub fn quota(&self) -> u64 {
582 self.memory_quota_bytes
583 }
584
585 /// Bytes currently accounted as in-use against the quota.
586 pub fn bytes_in_use(&self) -> u64 {
587 self.bytes_in_use.load(Ordering::Acquire)
588 }
589
590 /// Per-tenant GPU memory cap in bytes, or `None` for "no cap"
591 /// (operator-trust deployment).
592 ///
593 /// ADVISORY / IN-PROCESS-ONLY unless a [`DriverMemPool`] is wired in
594 /// via [`TenantContextBuilder::with_driver_enforced_gpu_cap`]. Set via
595 /// [`TenantContextBuilder::with_gpu_memory_bytes_cap`]. The in-process
596 /// allocator path
597 /// (`tensor-wasm-mem::TensorWasmMemoryCreator::with_tenant_context`)
598 /// reads this on every allocation and refuses to allocate when the
599 /// would-be new total of [`Self::gpu_bytes_in_use`] would exceed it.
600 /// Under `gpu-mem-pool`, the `TenantMemPool` *also* enforces the cap
601 /// host-side in its `allocate` method (a second line of defence for
602 /// pool-routed allocations). NB: there is NO driver-level per-pool
603 /// allocation ceiling — `CU_MEMPOOL_ATTR_RELEASE_THRESHOLD` is a
604 /// retention hint, not a cap (a hardware run disproved the earlier
605 /// driver-pin claim; see `docs/GPU-VALIDATION-2026-05-30.md` BUG-1). A
606 /// tenant calling the CUDA driver *directly* (not through TenantMemPool)
607 /// is therefore still uncapped; see `docs/GPU-QUOTAS.md`.
608 pub fn gpu_memory_bytes_cap(&self) -> Option<u64> {
609 self.gpu_memory_bytes_cap
610 }
611
612 /// Bytes currently accounted as in-use against the GPU cap.
613 ///
614 /// Mirrors [`Self::bytes_in_use`] for the GPU side of the quota.
615 /// Updated by [`Self::consume_gpu_bytes`] /
616 /// [`Self::release_gpu_bytes`].
617 pub fn gpu_bytes_in_use(&self) -> u64 {
618 self.gpu_bytes_in_use.load(Ordering::Acquire)
619 }
620
621 /// Atomically reserve `n` GPU bytes against the per-tenant cap.
622 ///
623 /// Returns `Err(TensorWasmError::GpuMemoryExhausted)` if
624 /// [`Self::gpu_memory_bytes_cap`] is set and the allocation would
625 /// push usage above it. When the cap is `None` ("no cap"), the
626 /// counter is still bumped (so dashboards and the per-tenant gauge
627 /// surface real utilisation) but the request is never refused. The
628 /// add is performed with `checked_add` so a malicious or buggy
629 /// caller cannot wrap the counter by repeatedly requesting close
630 /// to `u64::MAX` — the second such call observes the overflow and
631 /// returns `GpuMemoryExhausted` while leaving the counter unchanged.
632 ///
633 /// Mirrors `Self::consume_bytes_inner` for the GPU side; the
634 /// atomic discipline is intentionally identical so a single mental
635 /// model covers both counters.
636 ///
637 /// # v0.3.7 vs v0.4 contract
638 ///
639 /// This in-process counter is the primary enforcement: the allocator
640 /// path calls `consume_gpu_bytes` before handing back the buffer. A
641 /// tenant that bypasses the allocator but still routes through the
642 /// `gpu-mem-pool` `TenantMemPool` is *additionally* capped host-side
643 /// in `TenantMemPool::allocate` (a second line of defence wired via
644 /// [`TenantContextBuilder::with_driver_enforced_gpu_cap`]). NB: this
645 /// is NOT a driver-level cap — `CU_MEMPOOL_ATTR_RELEASE_THRESHOLD` is
646 /// a retention hint, not an allocation ceiling (see
647 /// `docs/GPU-QUOTAS.md` and `docs/GPU-VALIDATION-2026-05-30.md`
648 /// BUG-1). A tenant calling the CUDA driver *directly* is still
649 /// uncapped; that threat is out of scope today.
650 pub fn consume_gpu_bytes(&self, n: u64) -> Result<(), TensorWasmError> {
651 let limit = self.gpu_memory_bytes_cap;
652 let mut current = self.gpu_bytes_in_use.load(Ordering::Acquire);
653 loop {
654 // MEDIUM fix (uncapped overflow): the counter contract is
655 // "never refused when there is no cap". A `checked_add`
656 // overflow on the uncapped path used to surface as a
657 // non-retryable `GpuMemoryExhausted { limit: u64::MAX }`,
658 // violating that contract. When `limit` is `None` we instead
659 // saturate at `u64::MAX` and `warn!` — the request is always
660 // admitted. The overflow-as-error behaviour is retained only
661 // for the capped path (a saturating add there could mask an
662 // over-cap allocation), where `next > cap` already rejects it.
663 let next = match current.checked_add(n) {
664 Some(v) => v,
665 None => {
666 if limit.is_none() {
667 tracing::warn!(
668 target: "tensor_wasm_tenant::context",
669 tenant = %self.tenant_id,
670 current,
671 requested = n,
672 "consume_gpu_bytes overflow on uncapped tenant; saturating at u64::MAX",
673 );
674 u64::MAX
675 } else {
676 return Err(TensorWasmError::GpuMemoryExhausted {
677 requested: n,
678 limit: limit.unwrap_or(u64::MAX),
679 current,
680 });
681 }
682 }
683 };
684 if let Some(cap) = limit {
685 if next > cap {
686 return Err(TensorWasmError::GpuMemoryExhausted {
687 requested: n,
688 limit: cap,
689 current,
690 });
691 }
692 }
693 match self.gpu_bytes_in_use.compare_exchange_weak(
694 current,
695 next,
696 Ordering::AcqRel,
697 Ordering::Acquire,
698 ) {
699 Ok(_) => {
700 self.publish_gpu_memory_gauge(next);
701 // Pool cap alignment (T39): when a driver memory pool was
702 // wired in via `with_driver_enforced_gpu_cap` AND a cap is
703 // set, keep the pool's recorded cap aligned to this
704 // tenant's policy ceiling. NB: this is NOT what enforces
705 // the cap on the pool path — `set_release_threshold` sets
706 // `CU_MEMPOOL_ATTR_RELEASE_THRESHOLD`, a memory-RETENTION
707 // hint, not an allocation ceiling (see
708 // `docs/GPU-VALIDATION-2026-05-30.md` BUG-1). The pool's
709 // real cap is enforced host-side in
710 // `TenantMemPool::allocate`. We push the *cap* (a fixed
711 // policy ceiling), not the running total. A failure is
712 // logged but does NOT fail the consume: the in-process
713 // counter already accepted this allocation, and the pool
714 // cap is belt-and-braces.
715 if let (Some(pool), Some(cap)) = (&self.driver_mem_pool, limit) {
716 if let Err(e) = pool.set_release_threshold(cap) {
717 tracing::warn!(
718 target: "tensor_wasm_tenant::context",
719 tenant = %self.tenant_id,
720 cap,
721 error = %e,
722 "driver mem-pool set_release_threshold failed; \
723 in-process gpu cap still enforced",
724 );
725 }
726 }
727 return Ok(());
728 }
729 Err(observed) => current = observed,
730 }
731 }
732 }
733
734 /// Atomically release `n` GPU bytes back to the cap.
735 ///
736 /// Saturating on underflow — callers must not release more than
737 /// they consumed, but a bookkeeping mismatch is not fatal. Mirrors
738 /// `Self::release_bytes_inner` for the GPU side: CAS loop on
739 /// `gpu_bytes_in_use`, computing `saturating_sub` on each iteration.
740 /// The earlier `fetch_sub` + post-hoc clamp shape is intentionally
741 /// avoided here for the same reason described on the CPU sibling
742 /// method — a concurrent `consume_gpu_bytes` race must not be
743 /// silently erased by a clamping `store`.
744 pub fn release_gpu_bytes(&self, bytes: u64) {
745 let mut current = self.gpu_bytes_in_use.load(Ordering::Acquire);
746 let after = loop {
747 let next = current.saturating_sub(bytes);
748 match self.gpu_bytes_in_use.compare_exchange_weak(
749 current,
750 next,
751 Ordering::AcqRel,
752 Ordering::Acquire,
753 ) {
754 Ok(_) => {
755 if current < bytes {
756 tracing::warn!(
757 target: "tensor_wasm_tenant::context",
758 tenant = %self.tenant_id,
759 before = current,
760 bytes,
761 "release_gpu_bytes underflow clamped",
762 );
763 }
764 break next;
765 }
766 Err(observed) => current = observed,
767 }
768 };
769 self.publish_gpu_memory_gauge(after);
770 }
771
772 /// Capability-checked variant of [`Self::consume_gpu_bytes`].
773 ///
774 /// Performs the same `Self::check_capability` gate the CPU path uses
775 /// ([`Self::consume_bytes_with_capability`]) before touching the GPU
776 /// counter, so the GPU side gets the identical cross-tenant isolation
777 /// guarantee: a [`TenantCapability`] minted for a different tenant is
778 /// rejected with [`TensorWasmError::TenantIsolationViolation`] and the
779 /// `gpu_bytes_in_use` counter is left untouched. On a matching capability
780 /// the behaviour is exactly that of the uncapped [`Self::consume_gpu_bytes`]
781 /// (cap check, `checked_add`, optional driver-pin, metrics publish).
782 ///
783 /// Until this method landed, the GPU counter had *no* capability-gated
784 /// form (unlike `consume_bytes` / `consume_bytes_with_capability`); the
785 /// uncapped [`Self::consume_gpu_bytes`] is retained for the 0.3 line in
786 /// the same way the CPU path keeps its unchecked variant.
787 pub fn consume_gpu_bytes_with_capability(
788 &self,
789 cap: &TenantCapability,
790 n: u64,
791 ) -> Result<(), TensorWasmError> {
792 self.check_capability(cap, "quota.consume_gpu_bytes")?;
793 self.consume_gpu_bytes(n)
794 }
795
796 /// Capability-checked variant of [`Self::release_gpu_bytes`].
797 ///
798 /// Mirrors [`Self::release_bytes_with_capability`] on the GPU side:
799 /// returns [`TensorWasmError::TenantIsolationViolation`] if `cap` was
800 /// minted for a different tenant (leaving the counter untouched);
801 /// otherwise releases exactly as the uncapped [`Self::release_gpu_bytes`]
802 /// (saturating on underflow) and returns `Ok(())`. The public signature
803 /// is `Result` because the capability check is fallible, even though the
804 /// underflow path of the underlying release is a best-effort clamp.
805 pub fn release_gpu_bytes_with_capability(
806 &self,
807 cap: &TenantCapability,
808 bytes: u64,
809 ) -> Result<(), TensorWasmError> {
810 self.check_capability(cap, "quota.release_gpu_bytes")?;
811 self.release_gpu_bytes(bytes);
812 Ok(())
813 }
814
815 /// Attempt to admit a single operation against this tenant's
816 /// time-windowed rate limit.
817 ///
818 /// Convenience wrapper over [`Self::try_acquire_ops`] with `n == 1`. See
819 /// that method for the full contract.
820 pub fn try_acquire_op(&self) -> Result<(), RateLimited> {
821 self.try_acquire_ops(1)
822 }
823
824 /// Attempt to admit `n` operations against this tenant's time-windowed
825 /// rate limit.
826 ///
827 /// When no rate limit was configured (the default — see
828 /// [`TenantContextBuilder::with_rate_limit`]), this is an unconditional
829 /// `Ok(())`, preserving the historical pure byte-cap behaviour. When a
830 /// limit is configured, a monotonic-clock token bucket is refilled for
831 /// the wall-clock time elapsed since the last call and then asked for `n`
832 /// tokens; the call returns `Ok(())` and removes the tokens if at least
833 /// `n` are available, or [`RateLimited`] (leaving the bucket untouched)
834 /// otherwise. The admission is all-or-nothing, so a large request never
835 /// partially drains the bucket.
836 ///
837 /// This gate is orthogonal to the byte caps: it does not read or mutate
838 /// `bytes_in_use` / `gpu_bytes_in_use`. A typical scheduler calls
839 /// `try_acquire_op` per kernel launch (or `try_acquire_ops(bytes)` for a
840 /// bytes/sec budget) and, on `Err`, defers or sheds the request rather
841 /// than failing it permanently — unlike a byte-cap refusal, a rate-limit
842 /// refusal clears on its own once the bucket refills.
843 pub fn try_acquire_ops(&self, n: u64) -> Result<(), RateLimited> {
844 match &self.rate_limiter {
845 Some(bucket) => bucket.try_acquire_at(n, Instant::now()),
846 None => Ok(()),
847 }
848 }
849
850 /// Whether a time-windowed rate limit is configured for this tenant
851 /// (i.e. [`TenantContextBuilder::with_rate_limit`] was called). When
852 /// `false`, [`Self::try_acquire_op`] always admits.
853 pub fn has_rate_limit(&self) -> bool {
854 self.rate_limiter.is_some()
855 }
856
857 /// Atomically reserve `n` bytes against the quota.
858 ///
859 /// Returns `Err(TensorWasmError::MemoryExhausted)` if the allocation would push
860 /// usage above the configured quota; on success, [`Self::bytes_in_use`]
861 /// reflects the new total.
862 ///
863 /// The add is performed with `checked_add` so a tenant whose quota
864 /// is set to `u64::MAX` cannot wrap the counter by repeatedly
865 /// asking for `u64::MAX` bytes — the second such call observes the
866 /// overflow and returns `MemoryExhausted` while leaving the counter
867 /// pinned at `u64::MAX` (saturating).
868 ///
869 /// # Deprecated
870 ///
871 /// This unchecked variant cannot tell which tenant is doing the
872 /// mutation. Prefer [`Self::consume_bytes_with_capability`], which
873 /// requires a [`TenantCapability`] minted by
874 /// [`crate::TenantRegistry::register_with_capability`] and rejects
875 /// cross-tenant calls with [`TensorWasmError::TenantIsolationViolation`].
876 /// The unchecked form is retained for the 0.3 line and will be removed
877 /// in v0.4.
878 #[deprecated(
879 since = "0.3.6",
880 note = "use consume_bytes_with_capability; unchecked variant will be removed in v0.4"
881 )]
882 pub fn consume_bytes(&self, n: u64) -> Result<(), TensorWasmError> {
883 self.consume_bytes_inner(n)
884 }
885
886 /// Capability-checked variant of [`Self::consume_bytes`].
887 ///
888 /// Returns [`TensorWasmError::TenantIsolationViolation`] if `cap` was
889 /// minted for a different tenant; otherwise behaves exactly like the
890 /// (deprecated) unchecked variant. The check is a single integer
891 /// compare on the hot path — negligible compared to the CAS loop that
892 /// performs the actual quota arithmetic.
893 pub fn consume_bytes_with_capability(
894 &self,
895 cap: &TenantCapability,
896 n: u64,
897 ) -> Result<(), TensorWasmError> {
898 self.check_capability(cap, "quota.consume_bytes")?;
899 self.consume_bytes_inner(n)
900 }
901
902 /// Shared implementation: the lock-free CAS loop. Both the deprecated
903 /// `consume_bytes` and the checked `consume_bytes_with_capability`
904 /// delegate here so the atomic discipline lives in one place.
905 fn consume_bytes_inner(&self, n: u64) -> Result<(), TensorWasmError> {
906 let limit = self.memory_quota_bytes;
907 let mut current = self.bytes_in_use.load(Ordering::Acquire);
908 loop {
909 let next = match current.checked_add(n) {
910 Some(v) if v <= limit => v,
911 _ => {
912 return Err(TensorWasmError::MemoryExhausted {
913 requested: n,
914 limit,
915 });
916 }
917 };
918 match self.bytes_in_use.compare_exchange_weak(
919 current,
920 next,
921 Ordering::AcqRel,
922 Ordering::Acquire,
923 ) {
924 Ok(_) => {
925 self.publish_memory_gauge(next);
926 return Ok(());
927 }
928 Err(observed) => current = observed,
929 }
930 }
931 }
932
933 /// Verify that `cap` was minted for the same tenant this context
934 /// belongs to. Returns [`TensorWasmError::TenantIsolationViolation`]
935 /// labelled with the *capability's* tenant id (i.e. the offending
936 /// caller) and a `resource` string identifying the gated operation —
937 /// the offended tenant id is implicit in which context the call
938 /// landed on and is recorded by the surrounding span.
939 ///
940 /// H1 (unconditional registry binding): an additional check rejects
941 /// caps minted by a *different* registry (by `Arc::ptr_eq` on the
942 /// per-registry token allocations). The check is **fail-closed**:
943 /// a context that was never registered (no `registry_token` recorded)
944 /// cannot prove which registry — if any — vouches for the cap, so the
945 /// capability check is rejected outright rather than waved through.
946 /// This closes the prior fail-open hole where a builder-constructed
947 /// context (`registry_token == None`) plus a foreign cap naming the
948 /// same tenant id silently passed. Registry-minted contexts always
949 /// carry a token (stamped by
950 /// [`crate::TenantRegistry::register_with_capability`]) and continue
951 /// to pass against caps from the same registry.
952 ///
953 /// H1: this binding is enforced regardless of the `strict-cap-binding`
954 /// feature, so a release build without that feature still rejects
955 /// cross-registry capabilities and fails closed on unregistered
956 /// contexts.
957 fn check_capability(
958 &self,
959 cap: &TenantCapability,
960 resource: &'static str,
961 ) -> Result<(), TensorWasmError> {
962 if cap.tenant_id != self.tenant_id {
963 return Err(TensorWasmError::TenantIsolationViolation {
964 tenant_id: cap.tenant_id,
965 resource: resource.into(),
966 });
967 }
968 match self.registry_token.as_ref() {
969 // Registered context: the cap must come from the same
970 // registry. Report the cap's tenant id as the offender —
971 // the audit-flagged threat is a holder of one registry's
972 // cap using it against a namesake tenant in another
973 // registry.
974 Some(ctx_token) if std::sync::Arc::ptr_eq(ctx_token, &cap.registry_token) => {}
975 // Either the cap was minted by a different registry, or
976 // this context carries no registry provenance at all
977 // (builder-constructed, never registered). Fail closed in
978 // both cases.
979 _ => {
980 return Err(TensorWasmError::TenantIsolationViolation {
981 tenant_id: cap.tenant_id,
982 resource: resource.into(),
983 });
984 }
985 }
986 Ok(())
987 }
988
989 /// Push the current CPU-side `bytes_in_use` total into this tenant's
990 /// per-tenant CPU gauge series, if a metrics handle was wired into
991 /// this context at build time. Centralised so [`Self::consume_bytes`]
992 /// and [`Self::release_bytes`] share one update path. The `Gauge::set`
993 /// call is a single relaxed atomic store — cheap enough to live on the
994 /// allocation hot path.
995 ///
996 /// The CPU counter and the GPU counter report against distinct,
997 /// correctly-named per-tenant families and never collide on the same
998 /// labelled series: CPU usage owns
999 /// [`TensorWasmMetrics::cpu_memory_bytes_per_tenant`] here, while GPU
1000 /// usage owns [`TensorWasmMetrics::gpu_memory_bytes_per_tenant`] (see
1001 /// [`Self::publish_gpu_memory_gauge`]).
1002 fn publish_memory_gauge(&self, new_total: u64) {
1003 if let Some(metrics) = &self.metrics {
1004 metrics
1005 .cpu_memory_bytes_per_tenant()
1006 .get_or_create(&self.metrics_labels)
1007 .set(new_total);
1008 }
1009 }
1010
1011 /// Push the current `gpu_bytes_in_use` total into the per-tenant GPU
1012 /// gauge series.
1013 ///
1014 /// Centralised so [`Self::consume_gpu_bytes`] and
1015 /// [`Self::release_gpu_bytes`] share one update path. GPU usage is the
1016 /// owner of the correctly-named per-tenant family
1017 /// [`TensorWasmMetrics::gpu_memory_bytes_per_tenant`]; the CPU counter
1018 /// no longer writes to it (see [`Self::publish_memory_gauge`]), so the
1019 /// two no longer clobber each other last-write-wins.
1020 fn publish_gpu_memory_gauge(&self, new_total: u64) {
1021 if let Some(metrics) = &self.metrics {
1022 metrics
1023 .gpu_memory_bytes_per_tenant()
1024 .get_or_create(&self.metrics_labels)
1025 .set(new_total);
1026 }
1027 }
1028
1029 /// Recorded-only CUDA memory-pool release-threshold value.
1030 ///
1031 /// Returns `None` when the builder was not given an explicit value
1032 /// (the driver default applies), or `Some(bytes)` when set via
1033 /// [`TenantContextBuilder::with_recorded_cuda_mem_pool_quota`].
1034 ///
1035 /// ADVISORY / RECORD-ONLY: as the builder method's name indicates,
1036 /// the value is informational only and is NEVER enforced — the cust
1037 /// 0.3.x crate does not expose `cuMemPoolSetAttribute`, so the CUDA
1038 /// driver never sees this number. Enforcement of this crate's
1039 /// per-tenant quota lives entirely in [`Self::bytes_in_use`].
1040 pub fn cuda_mem_pool_quota_bytes(&self) -> Option<u64> {
1041 self.cuda_mem_pool_quota_bytes
1042 }
1043
1044 /// The driver-level memory pool wired into this tenant for
1045 /// driver-enforced GPU-cap enforcement, or `None` when no pool was
1046 /// provided at build time (the in-process `gpu_bytes_in_use` counter
1047 /// is then the only enforcement).
1048 ///
1049 /// Returned as the backend-agnostic
1050 /// [`tensor_wasm_core::mem_pool::DriverMemPool`] trait object: this
1051 /// crate never names the concrete `tensor-wasm-mem` pool type, which
1052 /// is what keeps the `mem` <-> `tenant` dependency graph acyclic.
1053 /// Set via [`TenantContextBuilder::with_driver_enforced_gpu_cap`].
1054 pub fn mem_pool(&self) -> Option<&Arc<dyn DriverMemPool>> {
1055 self.driver_mem_pool.as_ref()
1056 }
1057
1058 /// Push this tenant's CUDA context onto the calling thread's context
1059 /// stack, returning a RAII guard that pops it on drop. Returns `None`
1060 /// if the tenant has no `cust::context::Context` (i.e. either the
1061 /// `cuda` feature is disabled or `ContextIsolated` was not requested
1062 /// at build time).
1063 ///
1064 /// The guard borrows `&self`, so the `TenantContext` cannot be moved
1065 /// or dropped while the guard is live — the pop on drop is therefore
1066 /// guaranteed to pop *this* context, not someone else's that snuck
1067 /// onto the stack.
1068 #[cfg(feature = "cuda")]
1069 pub fn enter(&self) -> Option<CudaCtxGuard<'_>> {
1070 let ctx = self.cu_context.as_ref()?;
1071 CudaCtxGuard::push(ctx)
1072 .ok()
1073 .map(|g| g.with_tenant(self.tenant_id))
1074 }
1075
1076 #[cfg(not(feature = "cuda"))]
1077 /// No-op equivalent of [`Self::enter`] when the `cuda` feature is off.
1078 /// Always returns `None`.
1079 pub fn enter(&self) -> Option<CudaCtxGuard> {
1080 None
1081 }
1082
1083 /// Atomically release `n` bytes back to the quota. Saturating on
1084 /// underflow — callers must not release more than they consumed, but a
1085 /// bookkeeping mismatch is not fatal.
1086 ///
1087 /// Implemented as a CAS loop on `bytes_in_use`, computing
1088 /// `saturating_sub` on each iteration. The earlier
1089 /// `fetch_sub` + post-hoc `store(0)` shape was racy: between the
1090 /// `fetch_sub` and the clamp `store`, a concurrent
1091 /// [`Self::consume_bytes`] could CAS in a new value, and the
1092 /// unconditional `store(0)` would then erase that consume. With the
1093 /// CAS loop, the underflow-clamp path only writes when the value we
1094 /// underflowed on is still current; otherwise we retry against the
1095 /// observed value.
1096 ///
1097 /// # Deprecated
1098 ///
1099 /// This unchecked variant cannot tell which tenant is doing the
1100 /// mutation. Prefer [`Self::release_bytes_with_capability`], which
1101 /// requires a [`TenantCapability`] minted by
1102 /// [`crate::TenantRegistry::register_with_capability`] and rejects
1103 /// cross-tenant calls with [`TensorWasmError::TenantIsolationViolation`].
1104 /// The unchecked form is retained for the 0.3 line and will be removed
1105 /// in v0.4.
1106 #[deprecated(
1107 since = "0.3.6",
1108 note = "use release_bytes_with_capability; unchecked variant will be removed in v0.4"
1109 )]
1110 pub fn release_bytes(&self, bytes: u64) {
1111 self.release_bytes_inner(bytes);
1112 }
1113
1114 /// Capability-checked variant of [`Self::release_bytes`].
1115 ///
1116 /// Returns [`TensorWasmError::TenantIsolationViolation`] if `cap` was
1117 /// minted for a different tenant; otherwise behaves exactly like the
1118 /// (deprecated) unchecked variant. Returns `Ok(())` on success — the
1119 /// unchecked variant returns `()` because the underflow path is a
1120 /// best-effort clamp, but the capability check itself is fallible, so
1121 /// the public signature here is `Result<(), TensorWasmError>`.
1122 pub fn release_bytes_with_capability(
1123 &self,
1124 cap: &TenantCapability,
1125 bytes: u64,
1126 ) -> Result<(), TensorWasmError> {
1127 self.check_capability(cap, "quota.release_bytes")?;
1128 self.release_bytes_inner(bytes);
1129 Ok(())
1130 }
1131
1132 /// Shared implementation: CAS-loop `saturating_sub` + underflow warn.
1133 /// Both the deprecated `release_bytes` and the checked
1134 /// `release_bytes_with_capability` delegate here so the atomic
1135 /// discipline lives in one place.
1136 fn release_bytes_inner(&self, bytes: u64) {
1137 let mut current = self.bytes_in_use.load(Ordering::Acquire);
1138 let after = loop {
1139 let next = current.saturating_sub(bytes);
1140 match self.bytes_in_use.compare_exchange_weak(
1141 current,
1142 next,
1143 Ordering::AcqRel,
1144 Ordering::Acquire,
1145 ) {
1146 Ok(_) => {
1147 if current < bytes {
1148 tracing::warn!(
1149 target: "tensor_wasm_tenant::context",
1150 tenant = %self.tenant_id,
1151 before = current,
1152 bytes,
1153 "release_bytes underflow clamped",
1154 );
1155 }
1156 break next;
1157 }
1158 Err(observed) => current = observed,
1159 }
1160 };
1161 self.publish_memory_gauge(after);
1162 }
1163
1164 /// Whether this tenant owns a real `cust::context::Context`.
1165 ///
1166 /// Returns `true` only when the `cuda` feature is enabled **and**
1167 /// [`TenantContextBuilder::build`] successfully constructed a primary
1168 /// context for a `ContextIsolated` tenant. Callers that need a
1169 /// real CUDA context (rather than a stream-isolation downgrade) should
1170 /// check this before calling [`Self::enter`].
1171 pub fn has_real_context(&self) -> bool {
1172 #[cfg(feature = "cuda")]
1173 {
1174 self.cu_context.is_some()
1175 }
1176 #[cfg(not(feature = "cuda"))]
1177 {
1178 false
1179 }
1180 }
1181}
1182
1183/// RAII guard returned by [`TenantContext::enter`]: pushes the tenant's
1184/// CUDA context onto the calling thread's context stack on construction,
1185/// pops it on drop.
1186///
1187/// The lifetime ties the guard to its owning `TenantContext`, so the
1188/// context cannot be dropped (and the underlying primary context cannot
1189/// be released) while the guard is live. The pop is performed even if a
1190/// panic unwinds through the guard's scope — that's the whole point of
1191/// the RAII pattern here.
1192///
1193/// Note: cust 0.3.x exposes both the "new" primary-context API (which is
1194/// not stack-based) and a `legacy::ContextStack::push`/`pop` shim that
1195/// covers `cuCtxPushCurrent`/`cuCtxPopCurrent`. The trait
1196/// `cust::context::ContextHandle` is implemented for both the primary
1197/// `Context` and `legacy::UnownedContext`, so we can use the stack API
1198/// uniformly here.
1199#[cfg(feature = "cuda")]
1200pub struct CudaCtxGuard<'a> {
1201 // PhantomData ties the guard to the borrowing `TenantContext`.
1202 _ctx: std::marker::PhantomData<&'a cust::context::Context>,
1203 // Tenant id, used only in the `Drop` log if `cuCtxPopCurrent` fails.
1204 tenant_id: Option<TenantId>,
1205}
1206
1207#[cfg(feature = "cuda")]
1208impl<'a> CudaCtxGuard<'a> {
1209 /// Push `ctx` onto the calling thread's context stack.
1210 pub fn push(ctx: &'a cust::context::Context) -> Result<Self, cust::error::CudaError> {
1211 // ContextHandle is implemented for the primary `Context`, so the
1212 // legacy stack API accepts it directly. This matches the plan's
1213 // `cuCtxPushCurrent` requirement.
1214 cust::context::legacy::ContextStack::push(ctx)?;
1215 Ok(Self {
1216 _ctx: std::marker::PhantomData,
1217 tenant_id: None,
1218 })
1219 }
1220}
1221
1222#[cfg(feature = "cuda")]
1223impl<'a> CudaCtxGuard<'a> {
1224 /// Bind a tenant id to this guard so the `Drop` log can attribute pop
1225 /// failures back to the offending tenant.
1226 fn with_tenant(self, tenant_id: TenantId) -> Self {
1227 Self {
1228 tenant_id: Some(tenant_id),
1229 ..self
1230 }
1231 }
1232}
1233
1234#[cfg(feature = "cuda")]
1235impl Drop for CudaCtxGuard<'_> {
1236 fn drop(&mut self) {
1237 // Best-effort pop: if the stack is empty or another context was
1238 // pushed on top, we still pop the topmost context. Errors cannot
1239 // be returned from `Drop`; the next-best thing is a structured
1240 // log with the underlying CUDA error code and the tenant whose
1241 // guard tripped, so operators can correlate it with kernels in
1242 // flight at the time of the panic / scope exit.
1243 if let Err(e) = cust::context::legacy::ContextStack::pop() {
1244 tracing::error!(
1245 target: "tensor_wasm_tenant::context",
1246 error = ?e,
1247 tenant = ?self.tenant_id,
1248 "cuCtxPopCurrent failed in CudaCtxGuard::drop",
1249 );
1250 }
1251 }
1252}
1253
1254/// No-CUDA placeholder so the type name is callable from generic code
1255/// without requiring the caller to cfg-gate `Option<CudaCtxGuard>`.
1256#[cfg(not(feature = "cuda"))]
1257pub struct CudaCtxGuard;
1258
1259/// Builder for [`TenantContext`].
1260///
1261/// Fields default to a low-overhead, multi-tenant-safe configuration:
1262/// [`IsolationKind::StreamIsolated`], stream id `0`, and an 8 GiB memory
1263/// quota. Override with the chained `with_*` methods, then call
1264/// [`Self::build`].
1265#[derive(Debug)]
1266pub struct TenantContextBuilder {
1267 tenant_id: TenantId,
1268 isolation: IsolationKind,
1269 stream_id: u64,
1270 memory_quota_bytes: u64,
1271 cuda_mem_pool_quota_bytes: Option<u64>,
1272 /// See [`TenantContext::gpu_memory_bytes_cap`]. `None` (the default)
1273 /// keeps the historical "no cap" behaviour; set via
1274 /// [`TenantContextBuilder::with_gpu_memory_bytes_cap`].
1275 gpu_memory_bytes_cap: Option<u64>,
1276 /// See [`TenantContext::mem_pool`]. `None` (the default) leaves the
1277 /// in-process `gpu_bytes_in_use` counter as the only GPU-cap
1278 /// enforcement; set via
1279 /// [`TenantContextBuilder::with_driver_enforced_gpu_cap`].
1280 driver_mem_pool: Option<Arc<dyn DriverMemPool>>,
1281 /// `(ops_per_sec, burst)` for the time-windowed operation-rate limiter.
1282 /// `None` (the default) means no rate limit — see
1283 /// [`TenantContext::try_acquire_op`]. Set via
1284 /// [`TenantContextBuilder::with_rate_limit`]; materialised into a
1285 /// [`TokenBucket`] at [`Self::build`] time so the bucket's monotonic
1286 /// clock starts ticking when the context goes live, not when the builder
1287 /// was created.
1288 rate_limit: Option<(u64, u64)>,
1289 #[cfg(feature = "cuda")]
1290 cuda_device_index: Option<u32>,
1291 metrics: Option<TensorWasmMetrics>,
1292}
1293
1294impl TenantContextBuilder {
1295 /// Default quota: 8 GiB. Sized for a single H100 SXM partition under MPS.
1296 pub const DEFAULT_QUOTA_BYTES: u64 = 8 * 1024 * 1024 * 1024;
1297
1298 /// Create a builder with default isolation, stream id, and quota.
1299 pub fn new(tenant_id: TenantId) -> Self {
1300 Self {
1301 tenant_id,
1302 isolation: IsolationKind::default(),
1303 stream_id: 0,
1304 memory_quota_bytes: Self::DEFAULT_QUOTA_BYTES,
1305 cuda_mem_pool_quota_bytes: None,
1306 gpu_memory_bytes_cap: None,
1307 driver_mem_pool: None,
1308 rate_limit: None,
1309 #[cfg(feature = "cuda")]
1310 cuda_device_index: None,
1311 metrics: None,
1312 }
1313 }
1314
1315 /// Set the per-tenant GPU memory cap in bytes.
1316 ///
1317 /// `None` (the default) means no cap — the tenant's GPU memory
1318 /// usage is recorded on [`TenantContext::gpu_bytes_in_use`] but the
1319 /// allocator never refuses an over-cap request. Setting `Some(bytes)`
1320 /// makes the allocator path
1321 /// (`tensor-wasm-mem::TensorWasmMemoryCreator::with_tenant_context`)
1322 /// return [`tensor_wasm_core::error::TensorWasmError::GpuMemoryExhausted`]
1323 /// for any [`TenantContext::consume_gpu_bytes`] that would push the
1324 /// total above `bytes`.
1325 ///
1326 /// See `docs/GPU-QUOTAS.md` for the v0.3.7 record-only semantics and
1327 /// the v0.4 `cuMemPool` enforcement plan.
1328 pub fn with_gpu_memory_bytes_cap(mut self, bytes: u64) -> Self {
1329 self.gpu_memory_bytes_cap = Some(bytes);
1330 self
1331 }
1332
1333 /// Wire a shared [`TensorWasmMetrics`] registry into the context so
1334 /// every CPU-side [`TenantContext::consume_bytes`] /
1335 /// [`TenantContext::release_bytes`] transition updates the per-tenant
1336 /// series of [`TensorWasmMetrics::cpu_memory_bytes_per_tenant`] and
1337 /// every GPU-side [`TenantContext::consume_gpu_bytes`] /
1338 /// [`TenantContext::release_gpu_bytes`] transition updates the
1339 /// per-tenant series of
1340 /// [`TensorWasmMetrics::gpu_memory_bytes_per_tenant`]. The handle is
1341 /// cheap to clone (it shares an inner `Arc`); the caller normally
1342 /// passes the same registry the API gateway exposes via
1343 /// `GET /metrics`. Omitting this builder call (or passing `None`
1344 /// directly into a future fallible variant) leaves the tenant's
1345 /// memory accounting completely off the dashboard — useful for
1346 /// benches and standalone examples that do not run a Prometheus
1347 /// scrape.
1348 pub fn with_metrics(mut self, metrics: TensorWasmMetrics) -> Self {
1349 self.metrics = Some(metrics);
1350 self
1351 }
1352
1353 /// Override the isolation level.
1354 pub fn with_isolation(mut self, isolation: IsolationKind) -> Self {
1355 self.isolation = isolation;
1356 self
1357 }
1358
1359 /// Override the stream identifier.
1360 pub fn with_stream_id(mut self, stream_id: u64) -> Self {
1361 self.stream_id = stream_id;
1362 self
1363 }
1364
1365 /// Override the memory quota in bytes.
1366 pub fn with_memory_quota_bytes(mut self, memory_quota_bytes: u64) -> Self {
1367 self.memory_quota_bytes = memory_quota_bytes;
1368 self
1369 }
1370
1371 /// Record a CUDA memory-pool release-threshold value on the
1372 /// [`TenantContext`] **without** applying it to a real CUDA mem-pool.
1373 ///
1374 /// The honest name reflects what this method actually does today:
1375 /// because cust 0.3.x does not expose the `cuMemPool*` API surface
1376 /// (`cuMemPoolSetAttribute` / `CU_MEMPOOL_ATTR_RELEASE_THRESHOLD`),
1377 /// the value is stored on the context for inspection, metrics, and
1378 /// forward-compatibility — but the CUDA driver never sees it. The
1379 /// only enforcement of per-tenant memory usage in this crate is the
1380 /// in-process counter returned by [`TenantContext::bytes_in_use`]
1381 /// and the quota set via
1382 /// [`Self::with_memory_quota_bytes`]; allocations that bypass the
1383 /// `consume_bytes` / `release_bytes` pair are NOT capped at the
1384 /// CUDA-driver level.
1385 ///
1386 /// Upgrading cust (or going direct via
1387 /// `cuda::sys::cuMemPoolSetAttribute`) is tracked as a future-work
1388 /// item; when that lands, a new `with_cuda_mem_pool_quota` method
1389 /// will replace this one and the value will be applied to the
1390 /// driver's release threshold at build time.
1391 pub fn with_recorded_cuda_mem_pool_quota(mut self, bytes: u64) -> Self {
1392 self.cuda_mem_pool_quota_bytes = Some(bytes);
1393 self
1394 }
1395
1396 /// Wire a driver-level memory pool into this tenant so the per-tenant
1397 /// GPU cap is enforced by the CUDA driver itself, not just the
1398 /// in-process [`TenantContext::gpu_bytes_in_use`] counter.
1399 ///
1400 /// `pool` is any [`tensor_wasm_core::mem_pool::DriverMemPool`] — in
1401 /// production, `tensor-wasm-mem`'s
1402 /// `cuda_mem_pool::TenantMemPool` (backed by
1403 /// `cuMemPoolSetAttribute(CU_MEMPOOL_ATTR_RELEASE_THRESHOLD, ...)`);
1404 /// in tests, a mock. The parameter is the backend-agnostic trait
1405 /// object on purpose: taking the concrete `mem` type would require a
1406 /// `tenant -> mem` dependency, re-introducing the `mem` <-> `tenant`
1407 /// cycle that this whole abstraction exists to break.
1408 ///
1409 /// When set, [`TenantContext::consume_gpu_bytes`] pins the pool's
1410 /// release threshold to the cap configured by
1411 /// [`Self::with_gpu_memory_bytes_cap`] on every accepted allocation,
1412 /// so a tenant that obtained a raw CUDA driver handle and bypassed
1413 /// the in-process counter still hits a driver-level
1414 /// `CUDA_ERROR_OUT_OF_MEMORY` at the cap. Pair this with
1415 /// [`Self::with_gpu_memory_bytes_cap`]; without a cap the pool is
1416 /// stored but never pinned (there is no ceiling to push). See
1417 /// `docs/GPU-QUOTAS.md` for the threat model the driver pin closes.
1418 pub fn with_driver_enforced_gpu_cap(mut self, pool: Arc<dyn DriverMemPool>) -> Self {
1419 self.driver_mem_pool = Some(pool);
1420 self
1421 }
1422
1423 /// Enable a time-windowed operation-rate limiter on this tenant.
1424 ///
1425 /// `ops_per_sec` is the steady-state refill rate (operations admitted per
1426 /// second once the burst budget is spent); `burst` is the bucket depth —
1427 /// the maximum number of operations that can be admitted back-to-back
1428 /// before the steady-state rate gates further ones. The bucket starts
1429 /// full, so the first `burst` operations after `build()` are admitted
1430 /// immediately. Both arguments are clamped to a minimum of `1` internally
1431 /// so a `(0, 0)` configuration cannot wedge the tenant into a
1432 /// never-admits state.
1433 ///
1434 /// Use this to bound a noisy neighbour's scheduling footprint without
1435 /// touching its memory caps: a value such as `with_rate_limit(1000, 200)`
1436 /// admits short bursts of up to 200 kernel launches while holding the
1437 /// long-run average at 1000/s. The same primitive expresses a bytes/sec
1438 /// budget by passing the byte count to
1439 /// [`TenantContext::try_acquire_ops`] instead of `1`.
1440 ///
1441 /// **Default = no rate limit.** Omitting this call (the default) leaves
1442 /// [`TenantContext::try_acquire_op`] an unconditional `Ok(())`, preserving
1443 /// the historical pure high-water-mark byte-cap behaviour. The limiter is
1444 /// purely additive and orthogonal to the byte counters.
1445 pub fn with_rate_limit(mut self, ops_per_sec: u64, burst: u64) -> Self {
1446 self.rate_limit = Some((ops_per_sec, burst));
1447 self
1448 }
1449
1450 /// Set the CUDA device index this tenant's context should be built
1451 /// against. Only meaningful when the `cuda` feature is enabled and
1452 /// the isolation is `ContextIsolated`. Defaults to device 0.
1453 #[cfg(feature = "cuda")]
1454 pub fn with_cuda_device_index(mut self, device_index: u32) -> Self {
1455 self.cuda_device_index = Some(device_index);
1456 self
1457 }
1458
1459 /// Finalise into a `TenantContext`.
1460 ///
1461 /// If the builder requested [`IsolationKind::ContextIsolated`] but
1462 /// constructing the underlying `cust::context::Context` fails (for
1463 /// example, no CUDA device 0, MPS unavailable, OOM at primary-
1464 /// context retain), the resulting `TenantContext`'s isolation level
1465 /// is **downgraded** to [`IsolationKind::StreamIsolated`] so the
1466 /// reported isolation matches reality. Callers that need to
1467 /// distinguish "real context-isolated" from "downgraded" can call
1468 /// [`TenantContext::has_real_context`].
1469 #[allow(unused_mut)]
1470 pub fn build(mut self) -> TenantContext {
1471 #[cfg(feature = "cuda")]
1472 let cu_context = {
1473 let want_isolated = matches!(self.isolation, IsolationKind::ContextIsolated);
1474 let built = self.build_cuda_context();
1475 if want_isolated && built.is_none() {
1476 // Honest reporting: a ContextIsolated request that produced
1477 // no real `cust::context::Context` is actually stream-
1478 // isolated at the GPU level. Downgrade so `.isolation()`
1479 // does not lie to schedulers, dashboards, or auditors —
1480 // AND escalate visibility: operators who specified
1481 // `ContextIsolated` as a deployment constraint need to
1482 // know the driver could not honour it. We bump a
1483 // process-wide counter (read via
1484 // [`isolation_downgrade_count`]) and emit a structured
1485 // `error!` so the alert pipeline picks it up. The
1486 // per-failure-cause logs inside `build_cuda_context`
1487 // record the underlying CUDA error code.
1488 //
1489 // TODO(core): when `TensorWasmMetrics` grows
1490 // `isolation_downgrade_total()` (see `ISOLATION_DOWNGRADE_COUNT`
1491 // docs), also do `if let Some(m) = &self.metrics {
1492 // m.isolation_downgrade_total().inc(); }` here so the
1493 // downgrade is observable through the registry, not just
1494 // the process-local static.
1495 ISOLATION_DOWNGRADE_COUNT.fetch_add(1, Ordering::Relaxed);
1496 tracing::error!(
1497 target: "tensor_wasm_tenant::context",
1498 tenant = %self.tenant_id,
1499 requested = %IsolationKind::ContextIsolated,
1500 effective = %IsolationKind::StreamIsolated,
1501 "ContextIsolated requested but unavailable; downgraded to StreamIsolated",
1502 );
1503 self.isolation = IsolationKind::StreamIsolated;
1504 }
1505 built
1506 };
1507 #[cfg(not(feature = "cuda"))]
1508 let cu_context = ();
1509
1510 let metrics_labels = TenantLabels::new(self.tenant_id.to_string());
1511 TenantContext {
1512 tenant_id: self.tenant_id,
1513 isolation: self.isolation,
1514 stream_id: self.stream_id,
1515 memory_quota_bytes: self.memory_quota_bytes,
1516 bytes_in_use: AtomicU64::new(0),
1517 gpu_memory_bytes_cap: self.gpu_memory_bytes_cap,
1518 gpu_bytes_in_use: AtomicU64::new(0),
1519 cuda_mem_pool_quota_bytes: self.cuda_mem_pool_quota_bytes,
1520 driver_mem_pool: self.driver_mem_pool,
1521 rate_limiter: self
1522 .rate_limit
1523 .map(|(ops_per_sec, burst)| TokenBucket::new(ops_per_sec, burst)),
1524 cu_context,
1525 metrics: self.metrics,
1526 metrics_labels,
1527 // Stamped by `TenantRegistry::register_with_capability` after
1528 // the context is moved into the registry but before it is
1529 // Arc-wrapped; remains `None` for contexts the test harness
1530 // constructs without going through the registry. H1: always
1531 // present so registry binding is enforced unconditionally.
1532 registry_token: None,
1533 }
1534 }
1535
1536 /// Build the underlying primary `cust::context::Context` when the
1537 /// `cuda` feature is on AND the tenant requested `ContextIsolated`.
1538 /// Returns `None` for shared/stream-isolated tenants, or when device
1539 /// retain fails — caller (build()) then proceeds with a stub context
1540 /// and operator-visible logs make the degradation explicit.
1541 #[cfg(feature = "cuda")]
1542 fn build_cuda_context(&self) -> Option<cust::context::Context> {
1543 if !matches!(self.isolation, IsolationKind::ContextIsolated) {
1544 return None;
1545 }
1546 let device_idx = self.cuda_device_index.unwrap_or(0);
1547 let device = match cust::device::Device::get_device(device_idx as i32) {
1548 Ok(d) => d,
1549 Err(e) => {
1550 tracing::error!(
1551 target: "tensor_wasm_tenant::context",
1552 tenant = %self.tenant_id,
1553 device = device_idx,
1554 error = ?e,
1555 "Device::get_device failed; falling back to stream-isolated mode",
1556 );
1557 return None;
1558 }
1559 };
1560 match cust::context::Context::new(device) {
1561 Ok(ctx) => Some(ctx),
1562 Err(e) => {
1563 tracing::error!(
1564 target: "tensor_wasm_tenant::context",
1565 tenant = %self.tenant_id,
1566 device = device_idx,
1567 error = ?e,
1568 "Context::new failed; falling back to stream-isolated mode",
1569 );
1570 None
1571 }
1572 }
1573 }
1574}
1575
1576#[cfg(test)]
1577#[allow(deprecated)]
1578// These tests pre-date the capability gate and exercise the unchecked
1579// `consume_bytes` / `release_bytes` variants directly. The deprecation
1580// warning is the *signal* to callers — silencing it here is the only
1581// place it should be silenced, and only because these tests pin the
1582// shim's behaviour until the variants are removed in v0.4.
1583mod tests {
1584 use super::*;
1585 // Test-only: the mock `DriverMemPool` impl below returns this error type.
1586 // Kept out of the module-level imports so the non-test lib build doesn't
1587 // carry an unused import (the production paths use `DriverMemPool` only).
1588 use tensor_wasm_core::mem_pool::MemPoolError;
1589
1590 #[test]
1591 fn builder_defaults() {
1592 let ctx = TenantContext::builder(TenantId(1)).build();
1593 assert_eq!(ctx.id(), TenantId(1));
1594 assert_eq!(ctx.isolation(), IsolationKind::StreamIsolated);
1595 assert_eq!(ctx.stream_id(), 0);
1596 assert_eq!(ctx.quota(), TenantContextBuilder::DEFAULT_QUOTA_BYTES);
1597 assert_eq!(ctx.bytes_in_use(), 0);
1598 }
1599
1600 #[test]
1601 fn builder_overrides() {
1602 let ctx = TenantContext::builder(TenantId(7))
1603 .with_isolation(IsolationKind::ContextIsolated)
1604 .with_stream_id(42)
1605 .with_memory_quota_bytes(1024)
1606 .build();
1607 assert_eq!(ctx.isolation(), IsolationKind::ContextIsolated);
1608 assert_eq!(ctx.stream_id(), 42);
1609 assert_eq!(ctx.quota(), 1024);
1610 }
1611
1612 #[test]
1613 fn quota_consume_release_round_trip() {
1614 let ctx = TenantContext::builder(TenantId(2))
1615 .with_memory_quota_bytes(1024)
1616 .build();
1617 ctx.consume_bytes(256).unwrap();
1618 assert_eq!(ctx.bytes_in_use(), 256);
1619 ctx.consume_bytes(512).unwrap();
1620 assert_eq!(ctx.bytes_in_use(), 768);
1621 ctx.release_bytes(256);
1622 assert_eq!(ctx.bytes_in_use(), 512);
1623 }
1624
1625 #[test]
1626 fn quota_enforcement_rejects_over_limit() {
1627 let ctx = TenantContext::builder(TenantId(3))
1628 .with_memory_quota_bytes(1024)
1629 .build();
1630 ctx.consume_bytes(1000).unwrap();
1631 let err = ctx.consume_bytes(100).unwrap_err();
1632 match err {
1633 TensorWasmError::MemoryExhausted { requested, limit } => {
1634 assert_eq!(requested, 100);
1635 assert_eq!(limit, 1024);
1636 }
1637 other => panic!("expected MemoryExhausted, got {other:?}"),
1638 }
1639 // Failed allocation must not move the counter.
1640 assert_eq!(ctx.bytes_in_use(), 1000);
1641 }
1642
1643 #[test]
1644 fn release_saturates_on_underflow() {
1645 let ctx = TenantContext::builder(TenantId(4))
1646 .with_memory_quota_bytes(1024)
1647 .build();
1648 ctx.release_bytes(999); // never consumed; must not panic or wrap.
1649 assert_eq!(ctx.bytes_in_use(), 0);
1650 }
1651
1652 #[test]
1653 fn isolation_kind_names_are_stable() {
1654 assert_eq!(IsolationKind::Shared.name(), "shared");
1655 assert_eq!(IsolationKind::StreamIsolated.name(), "stream_isolated");
1656 assert_eq!(IsolationKind::ContextIsolated.name(), "context_isolated");
1657 }
1658
1659 #[test]
1660 fn isolation_kind_matches_each_variant() {
1661 for kind in [
1662 IsolationKind::Shared,
1663 IsolationKind::StreamIsolated,
1664 IsolationKind::ContextIsolated,
1665 ] {
1666 let ctx = TenantContext::builder(TenantId(99))
1667 .with_isolation(kind)
1668 .build();
1669 assert_eq!(ctx.isolation(), kind);
1670 // Display is the same as the name.
1671 assert_eq!(ctx.isolation().to_string(), kind.name());
1672 }
1673 }
1674
1675 #[test]
1676 fn isolation_kind_default_is_stream_isolated() {
1677 assert_eq!(IsolationKind::default(), IsolationKind::StreamIsolated);
1678 }
1679
1680 #[test]
1681 fn cuda_mem_pool_quota_default_is_none() {
1682 let ctx = TenantContext::builder(TenantId(5)).build();
1683 assert_eq!(ctx.cuda_mem_pool_quota_bytes(), None);
1684 }
1685
1686 #[test]
1687 fn cuda_mem_pool_quota_recorded() {
1688 let ctx = TenantContext::builder(TenantId(6))
1689 .with_recorded_cuda_mem_pool_quota(4 * 1024 * 1024 * 1024)
1690 .build();
1691 assert_eq!(
1692 ctx.cuda_mem_pool_quota_bytes(),
1693 Some(4 * 1024 * 1024 * 1024)
1694 );
1695 }
1696
1697 #[test]
1698 fn metrics_handle_absent_by_default_is_a_noop() {
1699 // No `with_metrics(...)` — the consume/release pair must continue
1700 // to work exactly as before. The point of this test is to pin the
1701 // backwards-compat contract: pre-existing call sites that do not
1702 // plumb a registry observe no behaviour change.
1703 let ctx = TenantContext::builder(TenantId(11))
1704 .with_memory_quota_bytes(8192)
1705 .build();
1706 ctx.consume_bytes(1024).unwrap();
1707 ctx.release_bytes(512);
1708 assert_eq!(ctx.bytes_in_use(), 512);
1709 }
1710
1711 #[test]
1712 fn metrics_handle_publishes_consume_and_release_totals() {
1713 let metrics = TensorWasmMetrics::new();
1714 let ctx = TenantContext::builder(TenantId(12))
1715 .with_memory_quota_bytes(1 << 20)
1716 .with_metrics(metrics.clone())
1717 .build();
1718
1719 // CPU consume/release publish to the per-tenant CPU family
1720 // `cpu_memory_bytes_per_tenant` (the CPU counterpart of the GPU
1721 // counter's family). See `publish_memory_gauge`.
1722 let labels = TenantLabels::new(TenantId(12).to_string());
1723 let cpu = || {
1724 metrics
1725 .cpu_memory_bytes_per_tenant()
1726 .get_or_create(&labels)
1727 .get()
1728 };
1729
1730 // Consume → the per-tenant CPU gauge reads the post-add value.
1731 ctx.consume_bytes(4096).unwrap();
1732 assert_eq!(cpu(), 4096);
1733
1734 // A second consume composes.
1735 ctx.consume_bytes(2048).unwrap();
1736 assert_eq!(cpu(), 6144);
1737
1738 // Release → the gauge reads the post-sub value.
1739 ctx.release_bytes(2048);
1740 assert_eq!(cpu(), 4096);
1741 }
1742
1743 #[test]
1744 fn gpu_metrics_publish_to_per_tenant_family() {
1745 // The GPU counter owns the per-tenant `gpu_memory_bytes_per_tenant`
1746 // family; the CPU counter no longer collides on it.
1747 let metrics = TensorWasmMetrics::new();
1748 let ctx = TenantContext::builder(TenantId(12))
1749 .with_metrics(metrics.clone())
1750 .build();
1751 let labels = TenantLabels::new(TenantId(12).to_string());
1752
1753 ctx.consume_gpu_bytes(4096).unwrap();
1754 assert_eq!(
1755 metrics
1756 .gpu_memory_bytes_per_tenant()
1757 .get_or_create(&labels)
1758 .get(),
1759 4096
1760 );
1761
1762 ctx.release_gpu_bytes(2048);
1763 assert_eq!(
1764 metrics
1765 .gpu_memory_bytes_per_tenant()
1766 .get_or_create(&labels)
1767 .get(),
1768 2048
1769 );
1770
1771 // A CPU consume on the same context must NOT perturb the GPU
1772 // per-tenant series — the collision is gone.
1773 ctx.consume_bytes(1024).unwrap();
1774 assert_eq!(
1775 metrics
1776 .gpu_memory_bytes_per_tenant()
1777 .get_or_create(&labels)
1778 .get(),
1779 2048
1780 );
1781 }
1782
1783 /// Mock [`DriverMemPool`] that records every `set_release_threshold`
1784 /// call so tests can assert the tenant cap is pushed through with the
1785 /// right value. Lives in the test module so it cannot leak into the
1786 /// public surface; it is the `tensor-wasm-mem`-free stand-in for
1787 /// `cuda_mem_pool::TenantMemPool` (which this crate must not depend
1788 /// on).
1789 #[derive(Debug, Default)]
1790 struct MockDriverMemPool {
1791 threshold: AtomicU64,
1792 set_calls: AtomicU64,
1793 /// When `true`, `set_release_threshold` returns an error to
1794 /// exercise the consume path's fail-soft behaviour.
1795 fail: bool,
1796 }
1797
1798 impl DriverMemPool for MockDriverMemPool {
1799 fn set_release_threshold(&self, bytes: u64) -> Result<(), MemPoolError> {
1800 self.set_calls.fetch_add(1, Ordering::SeqCst);
1801 if self.fail {
1802 return Err(MemPoolError::SetAttribute("mock forced failure".into()));
1803 }
1804 self.threshold.store(bytes, Ordering::SeqCst);
1805 Ok(())
1806 }
1807
1808 fn release_threshold(&self) -> Option<u64> {
1809 Some(self.threshold.load(Ordering::SeqCst))
1810 }
1811 }
1812
1813 #[test]
1814 fn driver_enforced_gpu_cap_pushes_threshold_on_consume() {
1815 let pool = Arc::new(MockDriverMemPool::default());
1816 let ctx = TenantContext::builder(TenantId(20))
1817 .with_gpu_memory_bytes_cap(4096)
1818 .with_driver_enforced_gpu_cap(pool.clone())
1819 .build();
1820
1821 // The accessor returns the wired pool.
1822 assert!(ctx.mem_pool().is_some());
1823
1824 // A successful GPU consume pins the driver threshold to the cap
1825 // (NOT the running total).
1826 ctx.consume_gpu_bytes(1000).unwrap();
1827 assert_eq!(pool.set_calls.load(Ordering::SeqCst), 1);
1828 assert_eq!(pool.threshold.load(Ordering::SeqCst), 4096);
1829 assert_eq!(pool.release_threshold(), Some(4096));
1830
1831 // A second consume re-pins to the same cap.
1832 ctx.consume_gpu_bytes(500).unwrap();
1833 assert_eq!(pool.set_calls.load(Ordering::SeqCst), 2);
1834 assert_eq!(pool.threshold.load(Ordering::SeqCst), 4096);
1835 }
1836
1837 #[test]
1838 fn driver_enforced_gpu_cap_not_pushed_without_cap() {
1839 // No `with_gpu_memory_bytes_cap` → no ceiling to push, so the
1840 // pool is stored but `set_release_threshold` is never called.
1841 let pool = Arc::new(MockDriverMemPool::default());
1842 let ctx = TenantContext::builder(TenantId(21))
1843 .with_driver_enforced_gpu_cap(pool.clone())
1844 .build();
1845 ctx.consume_gpu_bytes(1000).unwrap();
1846 assert_eq!(pool.set_calls.load(Ordering::SeqCst), 0);
1847 assert!(ctx.mem_pool().is_some());
1848 }
1849
1850 #[test]
1851 fn driver_enforced_gpu_cap_over_limit_does_not_push() {
1852 // An over-cap consume is rejected by the in-process counter
1853 // BEFORE the driver pin runs, so the pool sees no call for the
1854 // rejected allocation.
1855 let pool = Arc::new(MockDriverMemPool::default());
1856 let ctx = TenantContext::builder(TenantId(22))
1857 .with_gpu_memory_bytes_cap(1024)
1858 .with_driver_enforced_gpu_cap(pool.clone())
1859 .build();
1860 let err = ctx.consume_gpu_bytes(2048).unwrap_err();
1861 assert!(matches!(err, TensorWasmError::GpuMemoryExhausted { .. }));
1862 assert_eq!(pool.set_calls.load(Ordering::SeqCst), 0);
1863 // The rejected allocation must not move the counter either.
1864 assert_eq!(ctx.gpu_bytes_in_use(), 0);
1865 }
1866
1867 #[test]
1868 fn driver_enforced_gpu_cap_set_failure_is_fail_soft() {
1869 // If the driver pin fails, the in-process consume still succeeds:
1870 // the counter already accepted the allocation and the driver pin
1871 // is belt-and-braces.
1872 let pool = Arc::new(MockDriverMemPool {
1873 fail: true,
1874 ..Default::default()
1875 });
1876 let ctx = TenantContext::builder(TenantId(23))
1877 .with_gpu_memory_bytes_cap(4096)
1878 .with_driver_enforced_gpu_cap(pool.clone())
1879 .build();
1880 ctx.consume_gpu_bytes(1000).unwrap();
1881 assert_eq!(pool.set_calls.load(Ordering::SeqCst), 1);
1882 // Threshold never recorded because the mock errored before storing.
1883 assert_eq!(pool.threshold.load(Ordering::SeqCst), 0);
1884 // The in-process counter still reflects the accepted allocation.
1885 assert_eq!(ctx.gpu_bytes_in_use(), 1000);
1886 }
1887
1888 #[test]
1889 fn mem_pool_accessor_default_is_none() {
1890 let ctx = TenantContext::builder(TenantId(24)).build();
1891 assert!(ctx.mem_pool().is_none());
1892 }
1893
1894 #[test]
1895 fn metrics_two_tenants_produce_two_distinct_series() {
1896 // Mirrors the dashboard's expected shape: two registered tenants
1897 // reserving different amounts must surface as two distinct
1898 // labelled series in the Prometheus exposition. The per-tenant
1899 // family is the GPU counter's, so drive it via `consume_gpu_bytes`.
1900 let metrics = TensorWasmMetrics::new();
1901 let a = TenantContext::builder(TenantId(101))
1902 .with_metrics(metrics.clone())
1903 .build();
1904 let b = TenantContext::builder(TenantId(102))
1905 .with_metrics(metrics.clone())
1906 .build();
1907 a.consume_gpu_bytes(4096).unwrap();
1908 b.consume_gpu_bytes(8192).unwrap();
1909
1910 let text = metrics.encode_text();
1911 assert!(
1912 text.contains("tensor_wasm_gpu_memory_bytes_per_tenant{tenant_id=\"T#101\"} 4096"),
1913 "missing tenant 101 sample in:\n{text}"
1914 );
1915 assert!(
1916 text.contains("tensor_wasm_gpu_memory_bytes_per_tenant{tenant_id=\"T#102\"} 8192"),
1917 "missing tenant 102 sample in:\n{text}"
1918 );
1919 }
1920
1921 #[test]
1922 fn metrics_release_underflow_publishes_clamped_zero() {
1923 let metrics = TensorWasmMetrics::new();
1924 let ctx = TenantContext::builder(TenantId(13))
1925 .with_memory_quota_bytes(1 << 16)
1926 .with_metrics(metrics.clone())
1927 .build();
1928 // Underflow path: release without prior consume. The counter
1929 // clamps to zero and the per-tenant CPU gauge should reflect zero,
1930 // not wrap.
1931 let labels = TenantLabels::new(TenantId(13).to_string());
1932 ctx.release_bytes(123);
1933 assert_eq!(
1934 metrics
1935 .cpu_memory_bytes_per_tenant()
1936 .get_or_create(&labels)
1937 .get(),
1938 0
1939 );
1940 }
1941
1942 #[test]
1943 fn enter_returns_none_without_cuda_context() {
1944 // On the no-CUDA path `enter` always returns None; on CUDA-enabled
1945 // builds the default builder still produces a context-less tenant
1946 // (StreamIsolated) so the result is None either way without
1947 // explicit ContextIsolated + a real device.
1948 let ctx = TenantContext::builder(TenantId(8)).build();
1949 assert!(ctx.enter().is_none());
1950 }
1951
1952 #[test]
1953 fn release_underflow_does_not_overwrite_concurrent_consume() {
1954 // Regression test for the `fetch_sub` + unconditional `store(0)`
1955 // race: the old shape would race-erase a concurrent
1956 // `consume_bytes` between the `fetch_sub` and the clamping
1957 // `store`. With the CAS loop, the underflow-clamp only writes
1958 // when the value we observed underflowing is still current.
1959 //
1960 // Construction: pre-load the counter with `(BYTES - 1)` so the
1961 // releaser thread's `release_bytes(BYTES)` underflows on every
1962 // iteration. The consumer thread observes that underflow window
1963 // and races a `consume_bytes(CONSUME)` against it. After both
1964 // threads have run `ITERATIONS` times each, the final counter
1965 // must equal the algebraic sum (clamped to zero), regardless of
1966 // interleaving. The old implementation would drop consumes,
1967 // producing a final value that drifts below the expected one.
1968 use std::thread;
1969
1970 const ITERATIONS: u64 = 10_000;
1971 const BYTES: u64 = 100;
1972 const PRE_LOAD: u64 = BYTES - 1; // guarantees release underflows
1973 const CONSUME: u64 = 7;
1974
1975 let ctx = Arc::new(
1976 TenantContext::builder(TenantId(0xAFE))
1977 // Quota generous enough that consume_bytes never trips
1978 // the MemoryExhausted branch and skews the algebra.
1979 .with_memory_quota_bytes(u64::MAX)
1980 .build(),
1981 );
1982 // Pre-load to the underflow-edge.
1983 ctx.consume_bytes(PRE_LOAD).unwrap();
1984
1985 let releaser = {
1986 let ctx = Arc::clone(&ctx);
1987 thread::spawn(move || {
1988 for _ in 0..ITERATIONS {
1989 ctx.release_bytes(BYTES);
1990 }
1991 })
1992 };
1993 let consumer = {
1994 let ctx = Arc::clone(&ctx);
1995 thread::spawn(move || {
1996 for _ in 0..ITERATIONS {
1997 ctx.consume_bytes(CONSUME).unwrap();
1998 }
1999 })
2000 };
2001 releaser.join().expect("releaser thread panicked");
2002 consumer.join().expect("consumer thread panicked");
2003
2004 // Algebraic expectation, computed with saturating arithmetic so
2005 // each release that observed `current < BYTES` clamps to zero
2006 // rather than wrapping. We can't reconstruct the exact
2007 // interleaving here, but the upper and lower bounds bracket the
2008 // legitimate final value:
2009 // - Lower bound: every release underflows immediately, so
2010 // each one clamps the counter to zero before the consumer
2011 // re-adds its CONSUME. The final state is somewhere
2012 // between `0` (if a release ran last) and
2013 // `ITERATIONS * CONSUME` (if every consume ran after every
2014 // release). The post-condition we actually assert is
2015 // stronger: total consumes minus total clamped-releases is
2016 // bounded by the consumer's contribution.
2017 // - Upper bound: `PRE_LOAD + ITERATIONS * CONSUME`.
2018 let final_value = ctx.bytes_in_use();
2019 let upper = PRE_LOAD.saturating_add(ITERATIONS.saturating_mul(CONSUME));
2020 assert!(
2021 final_value <= upper,
2022 "final {final_value} exceeded upper bound {upper}"
2023 );
2024 // The critical invariant: with the old buggy `store(0)` the
2025 // consumer's contributions could be wholesale erased between
2026 // `fetch_sub` and `store`. With the CAS loop, every successful
2027 // `consume_bytes` either lands before or after a `release`
2028 // CAS, but is never silently overwritten. We assert that the
2029 // counter never went negative (u64 sentinel for that is the
2030 // wrap-around to near-MAX — anything in the high half of the
2031 // u64 range would signal the bug).
2032 assert!(
2033 final_value < u64::MAX / 2,
2034 "final {final_value} suggests wrap-around — the race was not fixed",
2035 );
2036 }
2037
2038 #[test]
2039 fn rate_limit_absent_by_default_always_admits() {
2040 // No `with_rate_limit` → the historical pure byte-cap behaviour:
2041 // `try_acquire_op` is an unconditional Ok and `has_rate_limit` is
2042 // false. Pins the backwards-compat contract.
2043 let ctx = TenantContext::builder(TenantId(30)).build();
2044 assert!(!ctx.has_rate_limit());
2045 for _ in 0..10_000 {
2046 ctx.try_acquire_op().expect("no limiter must always admit");
2047 }
2048 }
2049
2050 #[test]
2051 fn token_bucket_admits_up_to_burst_then_rejects() {
2052 // Deterministic, no sleep: drive the bucket at a fixed instant so no
2053 // refill happens between calls. A full bucket of depth 5 admits
2054 // exactly 5 ops, then rejects.
2055 let bucket = TokenBucket::new(/*ops_per_sec*/ 100, /*burst*/ 5);
2056 let t0 = Instant::now();
2057 for i in 0..5 {
2058 bucket
2059 .try_acquire_at(1, t0)
2060 .unwrap_or_else(|_| panic!("op {i} within burst must be admitted"));
2061 }
2062 let err = bucket
2063 .try_acquire_at(1, t0)
2064 .expect_err("6th op past burst must be rejected");
2065 assert_eq!(err.requested, 1);
2066 assert_eq!(err.available, 0);
2067 assert_eq!(err.ops_per_sec, 100);
2068 assert_eq!(err.burst, 5);
2069 }
2070
2071 #[test]
2072 fn token_bucket_refills_over_time() {
2073 // Inject elapsed time rather than sleeping: at 10 ops/s, one token
2074 // accrues every 100 ms. Drain the bucket, then advance the injected
2075 // clock and confirm refilled tokens are admitted.
2076 let bucket = TokenBucket::new(/*ops_per_sec*/ 10, /*burst*/ 2);
2077 let t0 = Instant::now();
2078 // Drain the initial burst of 2.
2079 bucket.try_acquire_at(1, t0).unwrap();
2080 bucket.try_acquire_at(1, t0).unwrap();
2081 assert!(bucket.try_acquire_at(1, t0).is_err(), "bucket drained");
2082
2083 // 100 ms later → exactly one token refilled.
2084 let t1 = t0 + Duration::from_millis(100);
2085 bucket
2086 .try_acquire_at(1, t1)
2087 .expect("one token should have refilled after 100ms");
2088 assert!(
2089 bucket.try_acquire_at(1, t1).is_err(),
2090 "only one token refilled; second must be rejected"
2091 );
2092
2093 // 1 s after t0 → bucket fully refilled, but capped at burst (2),
2094 // not 10. So exactly 2 admits then a reject.
2095 let t2 = t0 + Duration::from_secs(1);
2096 bucket.try_acquire_at(1, t2).unwrap();
2097 bucket.try_acquire_at(1, t2).unwrap();
2098 assert!(
2099 bucket.try_acquire_at(1, t2).is_err(),
2100 "refill is capped at burst depth"
2101 );
2102 }
2103
2104 #[test]
2105 fn token_bucket_acquire_n_is_all_or_nothing() {
2106 // A request larger than the available tokens is rejected without
2107 // partially draining the bucket.
2108 let bucket = TokenBucket::new(100, 5);
2109 let t0 = Instant::now();
2110 let err = bucket
2111 .try_acquire_at(8, t0)
2112 .expect_err("8 > burst 5 must reject");
2113 assert_eq!(err.requested, 8);
2114 assert_eq!(err.available, 5);
2115 // Nothing was removed — a subsequent in-budget request still sees the
2116 // full bucket.
2117 bucket
2118 .try_acquire_at(5, t0)
2119 .expect("full burst still available after a rejected over-budget request");
2120 }
2121
2122 #[test]
2123 fn try_acquire_ops_via_builder_uses_bytes_per_sec_budget() {
2124 // The same primitive expresses a bytes/sec budget: pass the byte
2125 // count to `try_acquire_ops`. burst=1000 bytes admits a 600+400
2126 // pair, then rejects a third that would exceed the bucket.
2127 let ctx = TenantContext::builder(TenantId(31))
2128 .with_rate_limit(/*bytes_per_sec*/ 1_000, /*burst*/ 1_000)
2129 .build();
2130 assert!(ctx.has_rate_limit());
2131 ctx.try_acquire_ops(600).unwrap();
2132 ctx.try_acquire_ops(400).unwrap();
2133 let err = ctx
2134 .try_acquire_ops(1)
2135 .expect_err("bucket drained to zero bytes");
2136 assert_eq!(err.burst, 1_000);
2137 }
2138
2139 #[test]
2140 fn rate_limit_zero_args_clamped_to_one() {
2141 // `with_rate_limit(0, 0)` must not wedge the tenant into never-admit:
2142 // both args clamp to 1, so the first op is admitted (full bucket of 1)
2143 // and the second is rejected at the same instant.
2144 let bucket = TokenBucket::new(0, 0);
2145 let t0 = Instant::now();
2146 bucket
2147 .try_acquire_at(1, t0)
2148 .expect("clamped burst of 1 admits one");
2149 assert!(bucket.try_acquire_at(1, t0).is_err());
2150 }
2151
2152 #[test]
2153 fn isolation_downgrade_counter_starts_at_zero() {
2154 // Reachable test for Fix B: the static counter is `0` at
2155 // startup. The downgrade path itself requires the `cuda`
2156 // feature AND a real-but-uncooperative CUDA device (or absence
2157 // of one) — neither is available in CI without hardware, so a
2158 // direct exercise of the downgrade branch is intentionally
2159 // omitted here. When the cust-or-cudarc upgrade lands and the
2160 // CUDA branch becomes mockable, replace this with a positive
2161 // assertion against `isolation_downgrade_count()` after
2162 // forcing a downgrade.
2163 //
2164 // NOTE: this assertion is order-sensitive. Other tests in this
2165 // module never call `TenantContextBuilder::build()` with
2166 // `IsolationKind::ContextIsolated` on a CUDA host, so the
2167 // counter stays at zero under `cargo test`. Under
2168 // `cargo test --features cuda` on a host without CUDA, this
2169 // test will observe the counter is non-zero — the test then
2170 // documents (rather than enforces) the downgrade contract.
2171 let count = isolation_downgrade_count();
2172 // The getter must be a pure read: calling it twice in a row
2173 // returns the same value (no side effect, no implicit reset).
2174 // This holds on every build matrix regardless of `cuda`.
2175 assert_eq!(
2176 count,
2177 isolation_downgrade_count(),
2178 "isolation_downgrade_count() must be a side-effect-free read",
2179 );
2180 #[cfg(not(feature = "cuda"))]
2181 {
2182 assert_eq!(
2183 count, 0,
2184 "isolation_downgrade_count should start at zero on no-CUDA builds; \
2185 a non-zero reading means a downgrade was attributed to the wrong \
2186 path or a prior test mutated the static",
2187 );
2188 }
2189 }
2190}