jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-31 14:31:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(relay): make cache admission and insertion one ordered operation

Browse Source 
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 <f4a42a97e594b77bdbd8ee35191c8b28a94a4cb871d96f32921558275421fb68@buzz.block.builderlab.xyz>
Signed-off-by: npub17jjz49l9jjmhhk7cac63j8yt9z555n9cw8vk7v5jz4vzw4ppld5qgj57cc <f4a42a97e594b77bdbd8ee35191c8b28a94a4cb871d96f32921558275421fb68@buzz.block.builderlab.xyz>
This commit is contained in:
npub17jjz49l9jjmhhk7cac63j8yt9z555n9cw8vk7v5jz4vzw4ppld5qgj57cc
2026-07-31 16:52:59 -04:00
parent c05cdb15d1
commit 7772018add
2 changed files with 299 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

110
crates/buzz-pubsub/src/community_topics.rs
View File

@@ -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<R>(&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<CommunityId>,
desired: &DesiredCommunities,
) -> Vec<CommunityId> {
let want = desired();
let withdrawn: Vec<CommunityId> = 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<CommunityId> {
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<Vec<CommunityId>> {
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<CommunityId> {
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.

249
crates/buzz-relay/src/state.rs
View File

@@ -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<Instant> {
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