From 7772018add15e15e18337c3ee277e2eba383343b Mon Sep 17 00:00:00 2001 From: npub17jjz49l9jjmhhk7cac63j8yt9z555n9cw8vk7v5jz4vzw4ppld5qgj57cc Date: Fri, 31 Jul 2026 16:52:59 -0400 Subject: [PATCH] fix(relay): make cache admission and insertion one ordered operation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review (Wren, ratified by Eva) found the same "an admitted insert must not outlive its protection" invariant failing at a second boundary. Round 1 closed the lock boundary; this is the lifetime boundary. `CacheResidency::admit()` stamped `deadline = now + AUTHZ_CACHE_TTL`, read establishment, and returned a bool. All three callers inserted into moka only afterwards, and moka starts its own identical TTL at that later instant. So the residency window that protects an entry started *before* the entry's own clock — under-covering it by the call gap at minimum, and by an unbounded amount if the thread is preempted in between. The lost interleaving: admit(C) renews residency to T+10s and returns true; the caller is delayed; after T+10s the reconciler prunes residency and correctly withdraws establishment and unsubscribes; the caller resumes and inserts an entry readable for nearly a full TTL with no invalidation channel behind it. The gate no longer hands out a bool for the caller to act on. `CommunityTopics::with_established(community, f)` runs `f` inside the establishment read lock, and `CacheResidency::admit_and_insert(community, insert)` performs three ordered steps in one call: 1. residency is recorded before the establishment read (round 1's invariant, unchanged): withdrawal re-reads desire under the matching write lock, so either this renewal is seen and the subscription survives, or the withdrawal already happened and nothing is admitted; 2. `insert()` runs inside the read lock, so no withdrawal can interleave between the gate and the entry landing; 3. the residency deadline is re-stamped after `insert()` returns and still inside that section, so it is based at or after the entry's own clock start and therefore outlives it. `admit()` is deleted rather than deprecated, so read-then-act is unrepresentable rather than warned against — the same structural exclusion that removed `remove_established` in round 1. The three call sites cache three different value types (membership, accessible-channels, channel-visibility), so the insert travels as a closure rather than making the API generic over the cache. TTL padding was rejected: it narrows the window without closing arbitrary descheduling. The tests assert the exclusion *structurally*, with no threads and no sleeps. An earlier draft of this commit parked a withdrawal thread and slept 50ms from inside the critical section. That is deterministic on correct code, but its *mutation bite* is not: under the unlocked mutation a late-scheduled withdrawer still loses the race and the test passes vacuously. Wren caught it pre-push. A probabilistic detector is not a deterministic test, so both concurrency tests were rewritten to assert `try_write()`/`WouldBlock` from inside the critical section — the lock state itself rather than a racing thread's outcome. `CommunityTopics::try_withdraw_undesired_for_test` is the seam for that; it shares `withdraw_locked` with the real `withdraw_undesired` so the twin cannot drift from the path it stands in for. `residency_outlives_the_entry_it_protects` previously compared TTL constants and was green while the bug was live — equal durations do not nest when their clocks start in the wrong order. It now records `Instant::now()` from inside the insert closure and asserts `deadline >= inserted_at + AUTHZ_CACHE_TTL`, the nesting relation itself. Mutation results, each re-run to confirm determinism rather than a lucky pass: * releasing the establishment lock before calling `f` (round 2's read-then-act, restored) fails `no_entry_can_land_after_its_subscription_is_withdrawn` and `with_established_holds_the_gate_for_the_whole_closure` on 5 of 5 runs, and leaves `residency_outlives_the_entry_it_protects` green; * dropping the step-3 re-stamp fails both relay tests on 3 of 3 runs. It bites `no_entry_can_land...` too because that test's post-insert assertion — the next withdrawal must withdraw *nothing* — is exactly what the re-stamp buys; the lock alone does not survive a residency that lapsed during the insert. So the lock and the re-stamp are separately load-bearing: the first mutation isolates the lock, the second shows the re-stamp is required even with the lock intact. Also resolves the `rustdoc::private_intra_doc_links` warning introduced in the previous commit: `cargo doc -p buzz-pubsub --no-deps` emitted 4 warnings at c05cdb15d and emits 3 here, the remainder pre-existing and unrelated. Co-authored-by: npub17jjz49l9jjmhhk7cac63j8yt9z555n9cw8vk7v5jz4vzw4ppld5qgj57cc Signed-off-by: npub17jjz49l9jjmhhk7cac63j8yt9z555n9cw8vk7v5jz4vzw4ppld5qgj57cc --- crates/buzz-pubsub/src/community_topics.rs | 110 ++++++++- crates/buzz-relay/src/state.rs | 249 ++++++++++++++++----- 2 files changed, 299 insertions(+), 60 deletions(-) diff --git a/crates/buzz-pubsub/src/community_topics.rs b/crates/buzz-pubsub/src/community_topics.rs index 5b87c7c03c..17670a03c8 100644 --- a/crates/buzz-pubsub/src/community_topics.rs +++ b/crates/buzz-pubsub/src/community_topics.rs @@ -9,9 +9,9 @@ //! connection control; local cache residency for cache invalidation), //! supplied as a closure and recomputed from scratch on every reconcile. This //! module never caches it. Crucially, the *withdrawal* direction re-reads it -//! under the same lock that removes establishment (see -//! [`CommunityTopics::withdraw_undesired`]), so interest renewed after the -//! diff was computed cannot have its subscription pulled out from under it. +//! under the same lock that removes establishment (`withdraw_undesired`), so +//! interest renewed after the diff was computed cannot have its subscription +//! pulled out from under it. //! * **established** — the communities Redis has acknowledged a `SUBSCRIBE` for //! on the *current* connection. A community enters only after the ack returns, //! and the whole set is cleared when the connection ends. @@ -70,8 +70,11 @@ impl CommunityTopics { } /// Whether Redis has acknowledged this community's subscription on the - /// current connection. This is the gate for anything that must not become - /// readable before its invalidation topic is established. + /// current connection. + /// + /// This is an *observation*, not a gate: it is true only until the lock is + /// released. Anything that must not become readable without a live + /// invalidation topic has to run inside [`Self::with_established`]. pub fn is_established(&self, community: CommunityId) -> bool { self.established .read() @@ -79,6 +82,31 @@ impl CommunityTopics { .contains(&community) } + /// Runs `f` with this community's establishment state, holding it against + /// withdrawal for the whole call. + /// + /// This is the gate for state whose readability depends on the + /// subscription — in the relay, an authorization cache insert. Withdrawal + /// takes the matching write lock (`withdraw_undesired`), so the two cannot + /// interleave: either `f` completes and the withdrawal decision that + /// follows re-reads desire and sees whatever interest `f` recorded, or the + /// withdrawal already happened and `f` is told `false`. Reading + /// establishment and then acting on it in a second step reopens that gap, + /// however small the gap looks. + /// + /// `f` must not call back into this `CommunityTopics`: the read lock is + /// held, so a nested write would deadlock and a nested read can too if a + /// writer is already queued. It may touch state the `desired` closure + /// reads — that closure only ever runs under the write lock, which excludes + /// this one, so there is no lock cycle. + pub fn with_established(&self, community: CommunityId, f: impl FnOnce(bool) -> R) -> R { + let established = self + .established + .read() + .expect("community topics lock poisoned"); + f(established.contains(&community)) + } + /// Number of established communities, for the connect log line and for /// tests asserting how much of a desired set has been acked. pub fn established_count(&self) -> usize { @@ -118,7 +146,8 @@ impl CommunityTopics { /// /// This is the ordering invariant the establishment gate rests on, and the /// reason the decision cannot be made from a snapshot taken earlier in the - /// reconcile. Interest renewal ([`CacheResidency::admit`] in the relay) + /// reconcile. Interest renewal (`CacheResidency::admit_and_insert` in the + /// relay) /// records residency and *then* reads establishment under the read lock; /// this method re-reads residency under the matching write lock. Mutual /// exclusion makes the two orderings the only possibilities: @@ -141,6 +170,15 @@ impl CommunityTopics { .established .write() .expect("community topics lock poisoned"); + Self::withdraw_locked(&mut established, desired) + } + + /// The withdrawal decision itself, given the write lock. Shared so the + /// non-blocking test twin below cannot drift away from the real path. + fn withdraw_locked( + established: &mut HashSet, + desired: &DesiredCommunities, + ) -> Vec { let want = desired(); let withdrawn: Vec = established.difference(&want).copied().collect(); for community in &withdrawn { @@ -149,13 +187,35 @@ impl CommunityTopics { withdrawn } - /// Test seam for [`Self::withdraw_undesired`], whose invariant is shared with + /// Test seam for `withdraw_undesired`, whose invariant is shared with /// the cache-residency gate in another crate. #[doc(hidden)] pub fn withdraw_undesired_for_test(&self, desired: &DesiredCommunities) -> Vec { self.withdraw_undesired(desired) } + /// Non-blocking `withdraw_undesired`: `None` when the establishment lock is + /// held elsewhere. + /// + /// Test seam for the exclusion itself. A test can call this from inside a + /// [`Self::with_established`] closure and assert `None` — structural proof + /// that no withdrawal can interleave with a gated insert, with no threads + /// and no scheduler dependence. Asserting that through a parked thread + /// makes the *mutation bite* probabilistic even though the passing case is + /// deterministic. + #[doc(hidden)] + pub fn try_withdraw_undesired_for_test( + &self, + desired: &DesiredCommunities, + ) -> Option> { + let mut established = match self.established.try_write() { + Ok(guard) => guard, + Err(std::sync::TryLockError::WouldBlock) => return None, + Err(e) => panic!("community topics lock poisoned: {e}"), + }; + Some(Self::withdraw_locked(&mut established, desired)) + } + fn snapshot(&self) -> HashSet { self.established .read() @@ -505,6 +565,42 @@ mod tests { ); assert!(!topics.is_established(c)); } + + /// `with_established` must exclude withdrawal for the whole closure, not + /// just for the establishment read. + /// + /// This is the lock half of the lifetime invariant, and it is asserted + /// *structurally*: from inside the closure, `try_write()` must report + /// `WouldBlock`. No threads, no sleeps, no scheduler. Under a mutation that + /// releases the lock before calling `f`, `try_write()` succeeds and this + /// fails on every run — which is what a deterministic mutation kill means. + /// An earlier draft signalled a withdrawal thread and slept 50ms to "give + /// it every chance"; that is deterministic on correct code but its + /// *mutation bite* depends on the scheduler, so it could pass vacuously. + /// Wren caught it pre-push. + #[test] + fn with_established_holds_the_gate_for_the_whole_closure() { + let topics = CommunityTopics::new("test"); + let c = community(1); + topics.insert_established(c); + + let established = topics.with_established(c, |established| { + assert!( + matches!( + topics.established.try_write(), + Err(std::sync::TryLockError::WouldBlock) + ), + "withdrawal must be excluded for the whole closure; a gate that \ + releases the lock before the caller acts is the round-2 bug" + ); + established + }); + + assert!(established); + // And the exclusion ends with the closure, or the reconciler would + // never be able to withdraw anything. + assert!(topics.established.try_write().is_ok()); + } } /// Redis-backed reconciliation tests. diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 380bc78321..8f34d655d5 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -176,43 +176,77 @@ impl CacheResidency { } } - /// Records that this pod is resolving authorization for `community` and - /// reports whether the result may be cached. + /// Records that this pod is resolving authorization for `community` and, if + /// the invalidation topic is established, runs `insert` to place the entry + /// in its cache. /// - /// Recording and gating are one call on purpose. Residency *is* the desired - /// set, so a community that is only gated and never recorded would never be - /// subscribed, never become established, and never pass the gate — the - /// authorization caches would stay off for the life of the pod. Folding the - /// two makes that ordering unrepresentable. + /// Recording, gating and inserting are one call on purpose, for two + /// separate reasons this lane found the hard way. + /// + /// **Recording with gating.** Residency *is* the desired set, so a community + /// that is only gated and never recorded would never be subscribed, never + /// become established, and never pass the gate — the authorization caches + /// would stay off for the life of the pod. Folding the two makes that + /// ordering unrepresentable. + /// + /// **Gating with inserting.** The residency window must cover the whole life + /// of every entry it protects. A gate that only returned a `bool` left the + /// caller to insert afterwards, so the moka entry's TTL started *after* the + /// residency deadline it was measured against — under-covering the entry by + /// the call gap, and by an unbounded amount if the thread is preempted in + /// between: residency could lapse, the reconciler withdraw and + /// `UNSUBSCRIBE`, and the resumed caller still land an entry readable for a + /// full TTL with nothing behind it. Equal TTL constants do not nest when the + /// clocks start in the wrong order. /// /// The first call for a community therefore returns `false` (nothing is /// acked yet) while still asking for the subscription; once the reconciler - /// establishes it, subsequent calls return `true`. A `false` means the pod - /// reads through to the DB — degraded, not stale. + /// establishes it, subsequent calls insert and return `true`. A `false` + /// means the pod reads through to the DB — degraded, not stale. /// /// Gating **inserts only** is deliberate: gating reads would discard a live, /// still-invalidatable cache on every reconnect and turn a pub/sub blip into /// a DB stampede. /// - /// **The order of the two statements below is load-bearing.** Residency is - /// recorded *before* establishment is read, and the reconciler withdraws - /// establishment while re-reading residency under the matching write lock - /// (`CommunityTopics::withdraw_undesired`). Those two facts together are - /// what stop an admitted entry from outliving its subscription: whichever - /// side wins the lock, either this renewal is visible to the withdrawal - /// decision (so the subscription survives) or the withdrawal already - /// happened (so `is_established` is false and nothing is admitted). Reading - /// establishment first, or withdrawing outside the lock, reopens the gap. - #[must_use] - pub fn admit(&self, community: CommunityId) -> bool { - let deadline = Instant::now() + self.ttl; - if self.resident.insert(community, deadline).is_none() { + /// **The three steps below are ordered, and each ordering is load-bearing.** + /// + /// 1. Residency is recorded *before* establishment is read, and the + /// reconciler withdraws establishment while re-reading residency under + /// the matching write lock (`CommunityTopics::withdraw_undesired`, + /// private to `buzz-pubsub`). + /// Whichever side wins the lock, either this renewal is visible to the + /// withdrawal decision (so the subscription survives) or the withdrawal + /// already happened (so nothing is admitted). + /// 2. `insert` runs *inside* [`CommunityTopics::with_established`], so no + /// withdrawal can interleave between the gate and the entry landing. + /// 3. The residency deadline is re-stamped *after* `insert` returns and + /// still inside that critical section, so it is based at or after the + /// entry's own clock start and therefore outlives it. + /// + /// `insert` must not call back into [`CommunityTopics`] (the read lock is + /// held); the three production callers do a single moka insert, which does + /// not. + pub fn admit_and_insert(&self, community: CommunityId, insert: impl FnOnce()) -> bool { + // Step 1: residency before the establishment read; see the ordering note. + if self + .resident + .insert(community, Instant::now() + self.ttl) + .is_none() + { // Newly resident: shorten the wait for the subscribe. Convergence // does not depend on this — the reconcile tick is the guarantee. self.topics.wake(); } - // Must follow the residency write above; see the ordering note. - let admitted = self.topics.is_established(community); + let admitted = self.topics.with_established(community, |established| { + if established { + // Step 2: the entry lands while withdrawal is excluded. + insert(); + // Step 3: residency restarts at or after the entry's clock, so + // the entry can never outlive the subscription protecting it. + self.resident.insert(community, Instant::now() + self.ttl); + } + established + }); if !admitted { metrics::counter!("buzz_authz_cache_insert_skipped_total").increment(1); } @@ -228,6 +262,26 @@ impl CacheResidency { self.resident.retain(|_, deadline| *deadline > now); self.resident.iter().map(|entry| *entry.key()).collect() } + + /// The deadline currently protecting `community`, if it is resident. + /// + /// Test seam: the nesting relation between a residency window and the cache + /// entry it protects is only checkable against the actual deadline. A test + /// comparing TTL *constants* passes while the clocks start in the wrong + /// order, which is exactly the bug this seam exists to catch. + #[doc(hidden)] + pub fn deadline_for_test(&self, community: CommunityId) -> Option { + self.resident.get(&community).map(|entry| *entry.value()) + } + + /// Forces every residency to lapse now. + /// + /// Test seam: lets a test reach the post-expiry state deterministically + /// instead of sleeping out a real TTL. + #[doc(hidden)] + pub fn expire_all_for_test(&self) { + self.resident.clear(); + } } /// Registers a socket, durably revalidates its community, then runs it. @@ -939,7 +993,8 @@ impl AppState { /// Check channel membership with a 10-second cache. Falls back to DB on miss. /// /// The result is cached only while this pod's cache-invalidation channel - /// for `community_id` is established — see [`CacheResidency::admit`]. + /// for `community_id` is established — see + /// [`CacheResidency::admit_and_insert`]. /// Ungated, a pod could cache an authorization decision it would never hear /// the invalidation for, and serve it for the whole TTL. pub async fn is_member_cached( @@ -955,9 +1010,9 @@ impl AppState { } metrics::counter!("buzz_membership_cache_misses_total").increment(1); let result = self.db.is_member(community_id, channel_id, pubkey).await?; - if self.cache_residency.admit(community_id) { + self.cache_residency.admit_and_insert(community_id, || { self.membership_cache.insert(key, result); - } + }); Ok(result) } @@ -1221,9 +1276,9 @@ impl AppState { .db .get_accessible_channel_ids(community_id, pubkey) .await?; - if self.cache_residency.admit(community_id) { + self.cache_residency.admit_and_insert(community_id, || { self.accessible_channels_cache.insert(key, result.clone()); - } + }); Ok(result) } @@ -1271,10 +1326,10 @@ impl AppState { // costs liveness rather than confidentiality — a stale `private` // is over-restrictive, not a leak — but a private->open flip going // unheard for the full TTL is still a real delivery bug. - if self.cache_residency.admit(community_id) { + self.cache_residency.admit_and_insert(community_id, || { self.channel_visibility_cache .insert((community_id, channel_id), visibility.clone()); - } + }); } Ok(visibility) } @@ -2066,40 +2121,114 @@ mod tests { } } - /// Residency drives the cache-invalidation desired set, so it must cover a - /// community for at least as long as an entry cached under it can be read. + /// Residency must cover an entry for at least as long as that entry can be + /// read — which is a statement about *nested lifetimes*, not about equal + /// constants. + /// + /// This test previously asserted `AUTHZ_CACHE_TTL == AUTHZ_CACHE_TTL` and + /// was green while the bug was live: `admit` stamped the residency deadline + /// and returned, and the caller started moka's identical TTL afterwards, so + /// the windows had the same length and the wrong offset. Equal durations do + /// not nest when their clocks start in the wrong order. It now asserts the + /// relation that actually matters: the residency deadline is taken *after* + /// the insert it protects. #[test] fn residency_outlives_the_entry_it_protects() { - // The two are the same constant by construction; asserting the relation - // is what fails the build if a future change gives the caches a longer - // TTL than residency, which would silently reopen the stale-authz gap. let topics = Arc::new(CommunityTopics::new("test")); let residency = CacheResidency::new(Arc::clone(&topics), AUTHZ_CACHE_TTL); let community = CommunityId::from_uuid(Uuid::from_u128(0xa)); + topics.insert_established_for_test(community); - let _ = residency.admit(community); - assert_eq!( - residency.resident_communities(), - HashSet::from([community]), - "a just-resolved community is resident" + // Stand in for the cache entry: record when its own TTL clock starts. + let inserted_at = Arc::new(std::sync::Mutex::new(None)); + let recorder = Arc::clone(&inserted_at); + assert!(residency.admit_and_insert(community, move || { + *recorder.lock().unwrap() = Some(Instant::now()); + })); + let inserted_at = inserted_at.lock().unwrap().expect("insert ran"); + + let deadline = residency.deadline_for_test(community).expect("resident"); + assert!( + deadline >= inserted_at + AUTHZ_CACHE_TTL, + "the residency window must start at or after the entry's own clock, \ + or the subscription can be withdrawn while the entry is still readable" ); // A residency shorter than the cache TTL is the bug; prove the tracker // reports lapse purely on its own clock by using a zero-length window. let expired = CacheResidency::new(topics, Duration::ZERO); - let _ = expired.admit(community); + expired.admit_and_insert(community, || {}); assert!( expired.resident_communities().is_empty(), "residency must lapse on its own timer, with no external event" ); } + /// The lifetime boundary of the same invariant: an entry cannot land after + /// its subscription has been withdrawn, because the insert runs while + /// withdrawal is structurally excluded. + /// + /// This is Wren's round-2 blocker, asserted without threads. From inside + /// the insert closure — the exact point where the old code had already + /// returned and left the caller unprotected — residency is lapsed, which is + /// the state that used to permit an unprotected entry, and a withdrawal is + /// then attempted. `try_withdraw_undesired_for_test` reports `WouldBlock`, + /// proving the reconciler cannot unsubscribe while this entry is landing. + /// After the call returns the same withdrawal both *acquires* the lock (so + /// the exclusion is scoped to the insert, not a lock this test wedged) and + /// withdraws nothing (so step 3's re-stamp put the entry back under cover). + /// + /// An earlier draft did this with a parked thread and a 50ms sleep. On + /// correct code that is deterministic, but its *mutation bite* was + /// scheduler-dependent — a late-scheduled withdrawer makes the unlocked + /// mutation pass. Wren caught that pre-push; a probabilistic detector is + /// not a deterministic test. This shape fails on every run under the + /// unlocked mutation. + #[test] + fn no_entry_can_land_after_its_subscription_is_withdrawn() { + let topics = Arc::new(CommunityTopics::new("test")); + let residency = Arc::new(CacheResidency::new(Arc::clone(&topics), AUTHZ_CACHE_TTL)); + let community = CommunityId::from_uuid(Uuid::from_u128(0xa)); + topics.insert_established_for_test(community); + + // Exactly the closure main.rs hands the cache-invalidation subscriber. + let for_closure = Arc::clone(&residency); + let desired: buzz_pubsub::community_topics::DesiredCommunities = + Arc::new(move || for_closure.resident_communities()); + + let lapse = Arc::clone(&residency); + let topics_for_insert = Arc::clone(&topics); + let desired_for_insert = Arc::clone(&desired); + let admitted = residency.admit_and_insert(community, move || { + // The delayed-caller state: residency has lapsed, so a reconciler + // reaching the withdrawal decision now *would* unsubscribe. + lapse.expire_all_for_test(); + assert!( + topics_for_insert + .try_withdraw_undesired_for_test(&desired_for_insert) + .is_none(), + "the reconciler must not be able to withdraw while an admitted \ + entry is still landing — that gap is the stale-authz hole" + ); + }); + + assert!(admitted, "an established community is admitted"); + let after = topics + .try_withdraw_undesired_for_test(&desired) + .expect("the exclusion must end with the insert, or the reconciler stalls"); + assert!( + after.is_empty(), + "the entry landed inside a residency window re-stamped around it, so \ + the very next withdrawal must leave its subscription alone" + ); + } + /// The bootstrap must close: recording residency and gating the insert are /// one call, so a community this pod reads becomes desired *before* it is /// admitted, and is admitted as soon as the reconciler acks. /// - /// Split into a `may_cache` gate plus a separate `touch`, only an admitted - /// community would ever be recorded — nothing would enter the desired set, + /// Split into a separate gate and record, only an admitted community would + /// ever be recorded — nothing would enter the desired set, /// nothing would be subscribed, and the authorization caches would stay off /// for the life of the pod. This test is the deadlock detector, so it /// asserts the *first* call is both refused and residency-forming. @@ -2109,10 +2238,16 @@ mod tests { let residency = CacheResidency::new(Arc::clone(&topics), AUTHZ_CACHE_TTL); let community = CommunityId::from_uuid(Uuid::from_u128(0xa)); + let inserted = Arc::new(AtomicBool::new(false)); + let flag = Arc::clone(&inserted); assert!( - !residency.admit(community), + !residency.admit_and_insert(community, move || flag.store(true, Ordering::SeqCst)), "nothing is cacheable before Redis acks the invalidation channel" ); + assert!( + !inserted.load(Ordering::SeqCst), + "a refused admission must not run the insert" + ); assert_eq!( residency.resident_communities(), HashSet::from([community]), @@ -2122,17 +2257,22 @@ mod tests { // Stand in for the reconciler acking the SUBSCRIBE it was just asked for. topics.insert_established_for_test(community); + let flag = Arc::clone(&inserted); assert!( - residency.admit(community), + residency.admit_and_insert(community, move || flag.store(true, Ordering::SeqCst)), "an acked community must be admitted" ); + assert!( + inserted.load(Ordering::SeqCst), + "an admitted community must actually get its entry inserted" + ); } - /// The cross-crate half of the ordering invariant: a community `admit()` - /// just made resident must survive a withdrawal decision taken concurrently - /// by the reconciler. + /// The cross-crate half of the ordering invariant: a community + /// `admit_and_insert()` just made resident must survive a withdrawal + /// decision taken concurrently by the reconciler. /// - /// `admit` records residency *before* reading establishment, and the + /// `admit_and_insert` records residency *before* reading establishment, and the /// reconciler re-reads residency under the write lock, so a renewal that got /// its residency in cannot be unsubscribed. This test pairs the two real /// call sites — the relay's residency closure and the pubsub withdrawal — @@ -2150,14 +2290,17 @@ mod tests { // Established and resident: the steady state. topics.insert_established_for_test(community); - assert!(residency.admit(community)); + assert!(residency.admit_and_insert(community, || {})); // Reconciler asks whether to withdraw. Residency is live, so it must not. assert!( topics.withdraw_undesired_for_test(&desired).is_empty(), "a resident community must never be unsubscribed" ); - assert!(residency.admit(community), "and it stays admissible"); + assert!( + residency.admit_and_insert(community, || {}), + "and it stays admissible" + ); } /// A newly resident community wakes the cache-invalidation subscriber, so @@ -2168,7 +2311,7 @@ mod tests { let residency = CacheResidency::new(Arc::clone(&topics), AUTHZ_CACHE_TTL); let community = CommunityId::from_uuid(Uuid::from_u128(0xa)); - let _ = residency.admit(community); + residency.admit_and_insert(community, || {}); tokio::time::timeout(Duration::from_millis(500), topics.wake_notified()) .await