Lines
100 %
Functions
Branches
use super::*;
use sqlx::postgres::PgPoolOptions;
/// A pool handle that never connects — enough to make a slot "ready" without a
/// database behind it.
fn lazy_pool() -> Pool<Postgres> {
PgPoolOptions::new()
.max_connections(TENANT_MAX_CONNECTIONS)
.connect_lazy("postgres://unused@localhost/unused")
.expect("a syntactically valid DSN never fails to build a lazy pool")
}
/// A registry that gives up on a residency permit quickly, so the exhaustion
/// path is testable without a ten-second wait.
fn impatient_registry() -> PoolRegistry {
PoolRegistry {
reserve_deadline: Duration::from_millis(50),
..PoolRegistry::new(Duration::from_secs(300))
/// Installs a user and drives its slot to ready, as a successful build would.
///
/// The slot handle is dropped, as `acquire_connection` drops its own when it
/// returns: holding one means "a caller is mid-acquire", which is a state the
/// registry deliberately refuses to evict under.
async fn ready_user(registry: &PoolRegistry) -> Uuid {
let user = Uuid::new_v4();
let slot = registry.install(user).expect("registry lock");
let permit = registry.reserve().await.expect("under the ceiling");
slot.set(Arc::new(TenantPool::new(lazy_pool(), permit)))
.map_err(|_| "slot was already set")
.expect("slot is freshly installed");
user
/// The live pool handle for a resident tenant, as a connection would hold it.
fn pool_of(registry: &PoolRegistry, user: Uuid) -> Arc<TenantPool> {
registry
.pools
.lock()
.expect("registry lock")
.get(&user)
.and_then(|entry| entry.cell.get().cloned())
.expect("tenant is resident and ready")
fn age(registry: &PoolRegistry, user: Uuid, by: Duration) {
let mut pools = registry.pools.lock().expect("registry lock");
let entry = pools.get_mut(&user).expect("user is resident");
entry.last_used = Instant::now() - by;
fn age_check(registry: &PoolRegistry, user: Uuid, by: Duration) {
entry.checked_at = Instant::now() - by;
fn resident(registry: &PoolRegistry) -> usize {
registry.pools.lock().expect("registry lock").len()
#[tokio::test]
async fn touch_reports_a_resident_pool_and_refreshes_its_idle_clock() {
let registry = impatient_registry();
let user = ready_user(®istry).await;
age(®istry, user, Duration::from_secs(120));
assert!(registry.touch(user).expect("registry lock").is_some());
let refreshed = registry
.expect("still resident")
.last_used;
assert!(
refreshed.elapsed() < Duration::from_secs(1),
"touch must reset the idle clock"
);
async fn touch_reports_nothing_for_a_user_with_no_pool() {
.touch(Uuid::new_v4())
.is_none()
async fn touch_sweeps_pools_that_have_gone_idle() {
assert_eq!(registry.idle_timeout(), Duration::from_secs(300));
let idle = ready_user(®istry).await;
age(®istry, idle, Duration::from_secs(600));
registry.touch(Uuid::new_v4()).expect("registry lock");
assert_eq!(
resident(®istry),
0,
"an expired pool must be swept, not left holding its permit"
registry.residency.available_permits(),
MAX_RESIDENT_POOLS,
"sweeping must return the permit with the pool"
async fn install_hands_back_the_existing_slot_when_two_callers_race() {
let first = registry.install(user).expect("registry lock");
let second = registry.install(user).expect("registry lock");
Arc::ptr_eq(&first, &second),
"the loser of the race must reuse the winner's slot, not overwrite it"
assert_eq!(resident(®istry), 1, "a race must not leave two entries");
async fn evict_lru_takes_the_oldest_ready_pool() {
let oldest = ready_user(®istry).await;
let newest = ready_user(®istry).await;
age(®istry, oldest, Duration::from_secs(60));
assert!(evict_lru(
&mut registry.pools.lock().expect("registry lock")
));
let pools = registry.pools.lock().expect("registry lock");
assert!(!pools.contains_key(&oldest));
assert!(pools.contains_key(&newest));
/// The permit comes back only when a pool's last connection drops, so evicting
/// a pool somebody is using frees nothing: the newcomer waits out the whole
/// deadline and fails anyway, and the victim builds a SECOND pool beside its
/// live one. The oldest pool is passed over while it is in use.
async fn evict_lru_passes_over_a_pool_that_is_still_in_use() {
// Stand in for a live `TenantConnection`, which anchors its pool by Arc.
let in_use = pool_of(®istry, oldest);
{
assert!(pools.contains_key(&oldest), "the busy pool must survive");
assert!(!pools.contains_key(&newest), "the idle one goes instead");
// Once the connection lets go it becomes the obvious victim again.
drop(in_use);
!registry
.contains_key(&oldest)
async fn evict_lru_skips_a_slot_that_is_still_building() {
let building = Uuid::new_v4();
registry.install(building).expect("registry lock");
assert!(!evict_lru(
.contains_key(&building)
async fn a_dropped_pool_returns_its_permit() {
MAX_RESIDENT_POOLS - 1
registry.pools.lock().expect("registry lock").remove(&user);
"the permit rides with the pool, so discarding one returns the other"
async fn an_evicted_pool_stays_usable_for_whoever_still_holds_it() {
let in_use = pool_of(®istry, user);
!in_use.pool().is_closed(),
"eviction must un-register a pool, never close one out from under a request"
MAX_RESIDENT_POOLS - 1,
"the permit stays spent while the pool is still in use"
async fn forget_drops_only_the_slot_it_was_given() {
let failed = registry.install(user).expect("registry lock");
// The failing caller's slot is replaced by a fresh one before it cleans up.
registry.forget(user, &failed).expect("registry lock");
let replacement = registry.install(user).expect("registry lock");
let current = pools.get(&user).expect("replacement must survive");
Arc::ptr_eq(¤t.cell, &replacement),
"a stale caller must not evict the generation that replaced it"
async fn residency_is_capped_and_reserving_past_it_evicts() {
let mut held = Vec::new();
for _ in 0..MAX_RESIDENT_POOLS {
held.push(ready_user(®istry).await);
assert_eq!(registry.residency.available_permits(), 0);
// Release the local handles so an evicted pool can actually be dropped.
held.clear();
let permit = registry.reserve().await.expect("eviction frees a permit");
"reserving past the ceiling evicts exactly one pool"
drop(permit);
async fn reserve_gives_up_rather_than_hanging_when_nothing_is_evictable() {
// Every slot is still building, so `evict_lru` refuses all of them.
let mut permits = Vec::new();
registry.install(Uuid::new_v4()).expect("registry lock");
permits.push(registry.reserve().await.expect("under the ceiling"));
matches!(registry.reserve().await, Err(DBError::PoolBudgetExhausted)),
"a newcomer must fail fast, not park forever on a permit nobody will release"
async fn a_reservation_never_evicts_a_pool_that_is_in_use() {
// Anchoring the pool is what a live `TenantConnection` does.
held.push(pool_of(®istry, user));
// Every pool is still checked out, so no eviction could return a permit
// anyway: the reservation must time out having destroyed nothing. Taking a
// victim here used to cost the tenant its pool AND leave the newcomer to
// fail regardless, because the permit rides the pool until its last
// connection drops.
assert!(matches!(
registry.reserve().await,
Err(DBError::PoolBudgetExhausted)
"a waiting newcomer must not close pools that are still in use"
async fn a_freshly_installed_tenant_is_not_immediately_rechecked() {
registry.install(user).expect("registry lock");
.claim_recheck(user, Duration::from_secs(30))
.expect("registry lock"),
"install reads the directory already; re-reading at once is waste"
async fn only_one_caller_per_interval_gets_to_recheck() {
age_check(®istry, user, Duration::from_secs(60));
"a stale status must be re-read by someone"
for _ in 0..5 {
"concurrent acquires must not each query the admin pool"
async fn a_failed_recheck_is_given_back_so_the_next_caller_retries() {
.release_recheck(user, Duration::from_secs(30))
.expect("registry lock");
"a read that failed must not buy silence for the rest of the interval"
async fn an_unknown_tenant_is_always_due() {
.claim_recheck(Uuid::new_v4(), Duration::from_secs(30))
"a tenant with no pool is about to build one, which reads the directory"
async fn releasing_an_unknown_tenant_is_harmless() {
.release_recheck(Uuid::new_v4(), Duration::from_secs(30))
.expect("a tenant evicted mid-check must not fail the release");
/// A slot whose build was cancelled leaves an empty cell nobody holds. It has
/// no permit to give back, so eviction rightly passes over it — but the sweep
/// must still reclaim it, or the registry grows by one entry per cancelled
/// request and never shrinks.
async fn the_sweep_reclaims_a_slot_whose_build_was_abandoned() {
let abandoned = Uuid::new_v4();
registry.install(abandoned).expect("registry lock");
age(®istry, abandoned, Duration::from_secs(600));
!evict_lru(&mut registry.pools.lock().expect("registry lock")),
"an empty slot holds no permit, so evicting it frees nothing"
"an abandoned, unheld slot must not outlive the idle timeout"
/// The other half: a slot somebody is still building must survive the sweep.
async fn the_sweep_leaves_a_build_in_flight_alone() {
let _held = registry.install(building).expect("registry lock");
age(®istry, building, Duration::from_secs(600));
1,
"the caller still building holds the slot"