diff --git a/client/src/event_subscription.rs b/client/src/event_subscription.rs index 0cb59f63..357bc518 100644 --- a/client/src/event_subscription.rs +++ b/client/src/event_subscription.rs @@ -44,7 +44,7 @@ use std::pin::Pin; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::task::{Context, Poll}; -use storage_primitives::BucketId; +use storage_primitives::{BucketId, Role}; use subxt::ext::scale_value::{self, At}; use subxt::{OnlineClient, PolkadotConfig}; use tokio::sync::mpsc; @@ -249,6 +249,26 @@ pub enum StorageEvent { block_number: u32, }, + // ======================================================================== + // Membership Events + // ======================================================================== + /// A member was added to a bucket, or an existing member's role changed. + MemberSet { + bucket_id: BucketId, + member: AccountId32, + role: Role, + block_hash: H256, + block_number: u32, + }, + + /// A member was removed from a bucket. + MemberRemoved { + bucket_id: BucketId, + member: AccountId32, + block_hash: H256, + block_number: u32, + }, + // ======================================================================== // Replica Events // ======================================================================== @@ -289,6 +309,8 @@ impl StorageEvent { StorageEvent::BucketCreated { bucket_id, .. } => Some(*bucket_id), StorageEvent::BucketFrozen { bucket_id, .. } => Some(*bucket_id), StorageEvent::BucketDeleted { bucket_id, .. } => Some(*bucket_id), + StorageEvent::MemberSet { bucket_id, .. } => Some(*bucket_id), + StorageEvent::MemberRemoved { bucket_id, .. } => Some(*bucket_id), StorageEvent::ReplicaSynced { bucket_id, .. } => Some(*bucket_id), _ => None, } @@ -340,6 +362,8 @@ impl StorageEvent { StorageEvent::BucketCreated { block_hash, .. } => *block_hash, StorageEvent::BucketFrozen { block_hash, .. } => *block_hash, StorageEvent::BucketDeleted { block_hash, .. } => *block_hash, + StorageEvent::MemberSet { block_hash, .. } => *block_hash, + StorageEvent::MemberRemoved { block_hash, .. } => *block_hash, StorageEvent::ReplicaSynced { block_hash, .. } => *block_hash, StorageEvent::Unknown { block_hash, .. } => *block_hash, } @@ -367,6 +391,8 @@ impl StorageEvent { StorageEvent::BucketCreated { block_number, .. } => *block_number, StorageEvent::BucketFrozen { block_number, .. } => *block_number, StorageEvent::BucketDeleted { block_number, .. } => *block_number, + StorageEvent::MemberSet { block_number, .. } => *block_number, + StorageEvent::MemberRemoved { block_number, .. } => *block_number, StorageEvent::ReplicaSynced { block_number, .. } => *block_number, StorageEvent::Unknown { block_number, .. } => *block_number, } @@ -516,7 +542,9 @@ impl EventFilter { | StorageEvent::AgreementEnded { .. } => self.include_agreements, StorageEvent::BucketCreated { .. } | StorageEvent::BucketFrozen { .. } - | StorageEvent::BucketDeleted { .. } => self.include_bucket_lifecycle, + | StorageEvent::BucketDeleted { .. } + | StorageEvent::MemberSet { .. } + | StorageEvent::MemberRemoved { .. } => self.include_bucket_lifecycle, StorageEvent::ProviderRegistered { .. } | StorageEvent::ProviderAddedToBucket { .. } | StorageEvent::PrimaryProviderRemoved { .. } @@ -1130,6 +1158,21 @@ impl EventParser for StorageProviderEventParser { block_number, }), + // ── Membership ──────────────────────────────────────────────────── + "MemberSet" => Some(StorageEvent::MemberSet { + bucket_id: scale_decode::field_u64(&fields, "bucket_id")?, + member: scale_decode::field_account(&fields, "member")?, + role: field_role(&fields, "role")?, + block_hash, + block_number, + }), + "MemberRemoved" => Some(StorageEvent::MemberRemoved { + bucket_id: scale_decode::field_u64(&fields, "bucket_id")?, + member: scale_decode::field_account(&fields, "member")?, + block_hash, + block_number, + }), + // ── Replicas ────────────────────────────────────────────────────── "ReplicaSynced" => Some(StorageEvent::ReplicaSynced { bucket_id: scale_decode::field_u64(&fields, "bucket_id")?, @@ -1222,6 +1265,21 @@ fn field_removal_reason(fields: &scale_value::Composite, name: &str) -> Str .unwrap_or_else(|| "Unknown".to_string()) } +/// Read a `Role`-shaped variant field. Returns `None` if missing, not a variant, or an +/// unrecognized variant name. +fn field_role(fields: &scale_value::Composite, name: &str) -> Option { + match fields + .at(name) + .and_then(scale_decode::variant_name)? + .as_str() + { + "Admin" => Some(Role::Admin), + "Writer" => Some(Role::Writer), + "Reader" => Some(Role::Reader), + _ => None, + } +} + // ============================================================================ // Tests // ============================================================================ @@ -1371,4 +1429,48 @@ mod tests { assert!(event.is_agreement_event()); assert!(!event.is_checkpoint_event()); } + + #[test] + fn test_member_set_helpers() { + let member = AccountId32::new([6u8; 32]); + let event = StorageEvent::MemberSet { + bucket_id: 9, + member: member.clone(), + role: Role::Writer, + block_hash: H256::repeat_byte(0x03), + block_number: 111, + }; + + assert_eq!(event.bucket_id(), Some(9)); + assert_eq!(event.block_hash(), H256::repeat_byte(0x03)); + assert_eq!(event.block_number(), 111); + match event { + StorageEvent::MemberSet { + member: m, role, .. + } => { + assert_eq!(m, member); + assert_eq!(role, Role::Writer); + } + _ => panic!("expected MemberSet"), + } + } + + #[test] + fn test_member_removed_helpers() { + let member = AccountId32::new([7u8; 32]); + let event = StorageEvent::MemberRemoved { + bucket_id: 10, + member: member.clone(), + block_hash: H256::repeat_byte(0x04), + block_number: 222, + }; + + assert_eq!(event.bucket_id(), Some(10)); + assert_eq!(event.block_hash(), H256::repeat_byte(0x04)); + assert_eq!(event.block_number(), 222); + match event { + StorageEvent::MemberRemoved { member: m, .. } => assert_eq!(m, member), + _ => panic!("expected MemberRemoved"), + } + } } diff --git a/provider-node/src/auth.rs b/provider-node/src/auth.rs index ae99b1db..827497db 100644 --- a/provider-node/src/auth.rs +++ b/provider-node/src/auth.rs @@ -38,6 +38,19 @@ struct CachedMembership { fetched_at: Instant, } +/// A bucket's cache slot: its membership (if any) plus a generation counter bumped on every +/// [`MembershipCache::invalidate`] call. +/// +/// The generation lets a chain fetch that was already in flight when an `invalidate()` landed +/// detect that its result is now stale and must not be persisted — without it, a fetch started +/// just before a revocation could still write the pre-revocation members into the cache after +/// the invalidation ran, silently resurrecting access for a full TTL. +#[derive(Debug, Clone, Default)] +struct CacheSlot { + membership: Option, + generation: u64, +} + /// Trait for resolving bucket membership (enables mocking in tests). #[async_trait::async_trait] pub trait MembershipResolver: Send + Sync { @@ -46,7 +59,7 @@ pub trait MembershipResolver: Send + Sync { /// Membership cache backed by chain queries via subxt. pub struct MembershipCache { - cache: DashMap, + cache: DashMap, ttl: Duration, resolver: Box, } @@ -67,40 +80,77 @@ impl MembershipCache { bucket_id: u64, account: &AccountId32, ) -> Result, String> { - // Check cache first - if let Some(entry) = self.cache.get(&bucket_id) { - if entry.fetched_at.elapsed() < self.ttl { - return Ok(find_role(&entry.members, account)); + // Check cache first, remembering the generation so the fetch below can detect an + // invalidate() that lands while it's in flight. + let generation_before = match self.cache.get(&bucket_id) { + Some(slot) => { + if let Some(cached) = &slot.membership { + if cached.fetched_at.elapsed() < self.ttl { + return Ok(find_role(&cached.members, account)); + } + } + slot.generation } - } + None => 0, + }; // Cache miss or stale — fetch from chain match self.resolver.fetch_members(bucket_id).await { Ok(members) => { let role = find_role(&members, account); - self.cache.insert( - bucket_id, - CachedMembership { - members, - fetched_at: Instant::now(), - }, - ); + let membership = CachedMembership { + members, + fetched_at: Instant::now(), + }; + // Only persist the fetch if the generation hasn't moved since we started — + // otherwise an invalidate() ran while we awaited the chain and this result is + // stale. and_modify/or_insert_with run under the same per-key lock as + // invalidate()'s, so there is no window between the check and the write. + self.cache + .entry(bucket_id) + .and_modify(|slot| { + if slot.generation == generation_before { + slot.membership = Some(membership.clone()); + } + }) + .or_insert_with(|| CacheSlot { + membership: Some(membership), + generation: generation_before, + }); Ok(role) } Err(e) => { // Stale-while-revalidate: serve stale data if chain is unreachable - if let Some(entry) = self.cache.get(&bucket_id) { - tracing::warn!( - "Chain unreachable for bucket {} membership, serving stale data: {}", - bucket_id, - e - ); - return Ok(find_role(&entry.members, account)); + if let Some(slot) = self.cache.get(&bucket_id) { + if let Some(cached) = &slot.membership { + tracing::warn!( + "Chain unreachable for bucket {} membership, serving stale data: {}", + bucket_id, + e + ); + return Ok(find_role(&cached.members, account)); + } } Err(e) } } } + + /// Drop the cached entry for `bucket_id`, if any, so the next lookup re-fetches from + /// chain, and bump its generation so a fetch already in flight cannot resurrect the + /// invalidated membership once it completes. + pub fn invalidate(&self, bucket_id: u64) { + self.cache + .entry(bucket_id) + .and_modify(|slot| { + slot.membership = None; + slot.generation = slot.generation.saturating_add(1); + }) + .or_insert_with(|| CacheSlot { + membership: None, + generation: 1, + }); + } } fn find_role(members: &[(AccountId32, Role)], account: &AccountId32) -> Option { @@ -491,4 +541,130 @@ mod tests { let unknown = AccountId32::new([4u8; 32]); assert_eq!(find_role(&members, &unknown), None); } + + /// Resolver that counts calls via a shared counter, so tests can assert whether a + /// lookup hit the cache or went to the resolver. + struct CountingResolver { + members: Vec<(AccountId32, Role)>, + calls: std::sync::Arc, + } + + #[async_trait::async_trait] + impl MembershipResolver for CountingResolver { + async fn fetch_members(&self, _bucket_id: u64) -> Result, String> { + self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Ok(self.members.clone()) + } + } + + #[tokio::test] + async fn test_invalidate_forces_refetch() { + let alice = AccountId32::new([1u8; 32]); + let calls = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let resolver = CountingResolver { + members: vec![(alice.clone(), Role::Admin)], + calls: calls.clone(), + }; + let cache = MembershipCache::new(Box::new(resolver), Duration::from_secs(300)); + + assert_eq!(cache.get_role(1, &alice).await, Ok(Some(Role::Admin))); + assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 1); + + // Second lookup within TTL hits the cache — no new resolver call. + assert_eq!(cache.get_role(1, &alice).await, Ok(Some(Role::Admin))); + assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 1); + + cache.invalidate(1); + + // Lookup after invalidation must re-fetch. + assert_eq!(cache.get_role(1, &alice).await, Ok(Some(Role::Admin))); + assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn test_invalidate_absent_bucket_is_noop() { + let calls = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let resolver = CountingResolver { + members: vec![], + calls: calls.clone(), + }; + let cache = MembershipCache::new(Box::new(resolver), Duration::from_secs(300)); + cache.invalidate(42); // no entry cached yet — must not panic + + // Invalidating an uncached bucket must not change get_role's behavior: the first + // lookup still triggers exactly one resolver call, same as any other cold miss. + let account = AccountId32::new([1u8; 32]); + assert_eq!(cache.get_role(42, &account).await, Ok(None)); + assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 1); + } + + /// Resolver whose `fetch_members` blocks until the test tells it to proceed, so a test + /// can deterministically interleave an `invalidate()` call in the middle of a fetch that + /// is already in flight — reproducing the lost-update race `get_role` must not have. + struct BlockingResolver { + members: Vec<(AccountId32, Role)>, + calls: std::sync::Arc, + /// Signaled by the test once the in-flight fetch should resume. + proceed: std::sync::Arc, + /// Signaled by `fetch_members` once it has started, so the test knows it's safe to + /// invalidate before the fetch returns. + started: std::sync::Arc, + } + + #[async_trait::async_trait] + impl MembershipResolver for BlockingResolver { + async fn fetch_members(&self, _bucket_id: u64) -> Result, String> { + self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + self.started.notify_one(); + self.proceed.notified().await; + Ok(self.members.clone()) + } + } + + #[tokio::test] + async fn test_invalidate_during_in_flight_fetch_discards_stale_result() { + let alice = AccountId32::new([1u8; 32]); + let calls = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let proceed = std::sync::Arc::new(tokio::sync::Notify::new()); + let started = std::sync::Arc::new(tokio::sync::Notify::new()); + let resolver = BlockingResolver { + members: vec![(alice.clone(), Role::Admin)], + calls: calls.clone(), + proceed: proceed.clone(), + started: started.clone(), + }; + let cache = std::sync::Arc::new(MembershipCache::new( + Box::new(resolver), + Duration::from_secs(300), + )); + + // Kick off a fetch for bucket 1 and let it block inside fetch_members — simulating a + // request that started reading membership just before a revocation lands. + let cache_clone = cache.clone(); + let alice_clone = alice.clone(); + let fetch_task = tokio::spawn(async move { cache_clone.get_role(1, &alice_clone).await }); + started.notified().await; + + // A revocation lands while the fetch above is still in flight. + cache.invalidate(1); + + // Let the stale fetch complete. + proceed.notify_one(); + let result = fetch_task.await.unwrap(); + assert_eq!( + result, + Ok(Some(Role::Admin)), + "the in-flight fetch's own read reflects what it actually fetched" + ); + + // The stale result must NOT have been cached: the next lookup re-fetches from chain + // instead of serving the discarded pre-revocation membership for a full TTL. + proceed.notify_one(); // arm the next fetch_members call to also proceed immediately + assert_eq!(cache.get_role(1, &alice).await, Ok(Some(Role::Admin))); + assert_eq!( + calls.load(std::sync::atomic::Ordering::SeqCst), + 2, + "a fetch that raced against invalidate() must not persist its stale result" + ); + } } diff --git a/provider-node/src/chain_state_coordinator.rs b/provider-node/src/chain_state_coordinator.rs index 61be1500..4561f552 100644 --- a/provider-node/src/chain_state_coordinator.rs +++ b/provider-node/src/chain_state_coordinator.rs @@ -15,6 +15,7 @@ //! `stake`, and all settings stay current — no field-patching, no partial //! updates, no second writer. +use crate::auth::MembershipCache; use crate::negotiate::NonceCounter; use crate::storage::{NonceStore, NullNonceStore}; use async_trait::async_trait; @@ -139,6 +140,9 @@ pub struct ChainStateCoordinator { chain_ws_url: String, provider_account: AccountId32, chain_state: Arc, + /// Invalidated (never populated) on membership-changing events. `None` when auth is + /// disabled, in which case the coordinator's membership-invalidation path no-ops. + membership_cache: Option>, } impl ChainStateCoordinator { @@ -146,11 +150,13 @@ impl ChainStateCoordinator { chain_ws_url: String, provider_account: AccountId32, chain_state: Arc, + membership_cache: Option>, ) -> Self { Self { chain_ws_url, provider_account, chain_state, + membership_cache, } } @@ -241,6 +247,7 @@ impl ChainStateCoordinator { self.process_provider_events(&chain, &parsed, block_number) .await; + self.invalidate_membership_cache(&parsed); } Ok(()) @@ -262,6 +269,14 @@ impl ChainStateCoordinator { ) .await; } + + /// Invalidate the membership cache for any bucket whose membership changed in + /// `parsed`. No-op when auth is disabled (`membership_cache` is `None`). + fn invalidate_membership_cache(&self, parsed: &[StorageEvent]) { + if let Some(cache) = &self.membership_cache { + invalidate_membership_on_events(cache, parsed); + } + } } // ── state synchronisation (chain-client-agnostic) ───────────────────────────── @@ -417,6 +432,24 @@ pub fn is_relevant_provider_event(event: &StorageEvent, provider_account: &Accou } } +/// Invalidate the membership cache entry for any bucket whose membership changed in +/// `events`. Independent of [`is_relevant_provider_event`]: membership events are keyed by +/// bucket, not by provider account, so they apply regardless of whose block this is. +/// Invalidating a bucket that isn't cached is a no-op. +pub fn invalidate_membership_on_events(cache: &MembershipCache, events: &[StorageEvent]) { + for event in events { + let bucket_id = match event { + StorageEvent::MemberSet { bucket_id, .. } + | StorageEvent::MemberRemoved { bucket_id, .. } + | StorageEvent::BucketDeleted { bucket_id, .. } => Some(*bucket_id), + _ => None, + }; + if let Some(bucket_id) = bucket_id { + cache.invalidate(bucket_id); + } + } +} + // ── ChainStateCoordinatorHandle ─────────────────────────────────────────────── /// Keeps the coordinator alive. Drop or call [`stop`](Self::stop) to shut down. @@ -520,4 +553,93 @@ mod tests { }); assert_eq!(cs.constants.read().as_ref().unwrap().request_timeout, 100); } + + /// Resolver that counts calls via a shared counter, so tests can tell whether a + /// lookup hit the cache or went to the resolver. + struct CountingResolver { + calls: Arc, + } + + #[async_trait] + impl crate::auth::MembershipResolver for CountingResolver { + async fn fetch_members( + &self, + _bucket_id: u64, + ) -> Result, String> { + self.calls.fetch_add(1, Ordering::SeqCst); + Ok(vec![]) + } + } + + fn counting_cache() -> (MembershipCache, Arc) { + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let resolver = CountingResolver { + calls: calls.clone(), + }; + ( + MembershipCache::new(Box::new(resolver), Duration::from_secs(300)), + calls, + ) + } + + fn member_removed(bucket_id: u64) -> StorageEvent { + StorageEvent::MemberRemoved { + bucket_id, + member: AccountId32::new([9u8; 32]), + block_hash: H256::zero(), + block_number: 1, + } + } + + #[tokio::test] + async fn membership_event_invalidates_matching_bucket() { + let (cache, calls) = counting_cache(); + let account = AccountId32::new([1u8; 32]); + + cache.get_role(1, &account).await.unwrap(); + assert_eq!(calls.load(Ordering::SeqCst), 1); + + invalidate_membership_on_events(&cache, &[member_removed(1)]); + + cache.get_role(1, &account).await.unwrap(); + assert_eq!( + calls.load(Ordering::SeqCst), + 2, + "a membership event for the cached bucket must force a re-fetch" + ); + } + + #[tokio::test] + async fn unrelated_event_does_not_invalidate() { + let (cache, calls) = counting_cache(); + let account = AccountId32::new([1u8; 32]); + + cache.get_role(1, &account).await.unwrap(); + assert_eq!(calls.load(Ordering::SeqCst), 1); + + // A provider-lifecycle event carries no bucket_id and must not touch the cache, + // independent of is_relevant_provider_event. + let events = [StorageEvent::ProviderRegistered { + provider: account.clone(), + stake: 0, + block_hash: H256::zero(), + block_number: 1, + }]; + invalidate_membership_on_events(&cache, &events); + + cache.get_role(1, &account).await.unwrap(); + assert_eq!( + calls.load(Ordering::SeqCst), + 1, + "an unrelated event must not invalidate a cached bucket" + ); + } + + #[test] + fn invalidate_on_uncached_bucket_is_noop() { + let (cache, calls) = counting_cache(); + // Nothing cached yet — must not panic. + invalidate_membership_on_events(&cache, &[member_removed(42)]); + assert_eq!(calls.load(Ordering::SeqCst), 0); + } } diff --git a/provider-node/src/command.rs b/provider-node/src/command.rs index b06ed715..8fe5a274 100644 --- a/provider-node/src/command.rs +++ b/provider-node/src/command.rs @@ -167,7 +167,8 @@ fn configure_state(state: ProviderState, cli: &Cli) -> ProviderState { } /// Start the chain-state coordinator, which keeps `chain_state.current_block` -/// and `chain_state.provider_info` in sync with the chain. +/// and `chain_state.provider_info` in sync with the chain, and invalidates +/// `state.membership_cache` entries for buckets whose membership changes on-chain. /// /// Returns `None` only when the provider id isn't a valid account. The /// coordinator itself never fails to start: it connects in the background and @@ -192,6 +193,7 @@ fn start_chain_state_coordinator( cli.rpc.chain_rpc.clone(), provider_account, state.chain_state.clone(), + state.membership_cache.clone(), ); tracing::info!("Chain-state coordinator started (retries until the chain is reachable)"); diff --git a/provider-node/src/lib.rs b/provider-node/src/lib.rs index 32a5a65f..f6fc5458 100644 --- a/provider-node/src/lib.rs +++ b/provider-node/src/lib.rs @@ -32,9 +32,9 @@ pub mod types; pub use api::create_router; pub use chain_state_coordinator::{ - is_relevant_provider_event, refresh_if_relevant_event, refresh_provider_state, sync_constants, - ChainState, ChainStateChainClient, ChainStateCoordinator, ChainStateCoordinatorHandle, - PalletConstants, + invalidate_membership_on_events, is_relevant_provider_event, refresh_if_relevant_event, + refresh_provider_state, sync_constants, ChainState, ChainStateChainClient, + ChainStateCoordinator, ChainStateCoordinatorHandle, PalletConstants, }; pub use challenge_responder::{ ChallengeChainClient, ChallengeResponder, ChallengeResponderConfig, ChallengeResponderHandle, diff --git a/provider-node/tests/chain_state_integration.rs b/provider-node/tests/chain_state_integration.rs index bcc4ebee..fd063993 100644 --- a/provider-node/tests/chain_state_integration.rs +++ b/provider-node/tests/chain_state_integration.rs @@ -18,6 +18,12 @@ //! Pointed at an unreachable chain it must stay up, never panic, leave //! [`ChainState`] at its defaults (so `/negotiate` keeps returning 503), and //! shut down cleanly when stopped. +//! +//! 4. **Membership invalidation.** [`invalidate_membership_on_events`] scans a +//! finalized block's events for membership changes and evicts the affected +//! bucket from [`MembershipCache`] — keyed by `bucket_id`, independent of +//! [`is_relevant_provider_event`] — so the next `require_role` lookup re-fetches +//! fresh roles instead of serving a stale cached role until TTL expiry. use async_trait::async_trait; use sp_core::H256; @@ -27,10 +33,12 @@ use std::sync::Arc; use std::time::Duration; use storage_client::discovery::ProviderInfo; use storage_client::{ClientError, ProviderSettings, StorageEvent}; +use storage_primitives::Role; +use storage_provider_node::auth::{MembershipCache, MembershipResolver}; use storage_provider_node::{ - is_relevant_provider_event, refresh_if_relevant_event, refresh_provider_state, sync_constants, - ChainState, ChainStateChainClient, ChainStateCoordinator, NonceCounter, NonceStore, - PalletConstants, + invalidate_membership_on_events, is_relevant_provider_event, refresh_if_relevant_event, + refresh_provider_state, sync_constants, ChainState, ChainStateChainClient, + ChainStateCoordinator, NonceCounter, NonceStore, PalletConstants, }; /// A WS URL that refuses immediately: port 1 on loopback is never listening, so @@ -70,6 +78,7 @@ async fn coordinator_leaves_state_at_defaults_while_chain_unreachable() { UNREACHABLE_CHAIN.to_string(), provider_account(), chain_state.clone(), + None, ); let handle = coordinator.start(); @@ -96,6 +105,7 @@ async fn coordinator_shares_chain_state_with_caller() { UNREACHABLE_CHAIN.to_string(), provider_account(), chain_state.clone(), + None, ); let handle = coordinator.start(); @@ -117,6 +127,7 @@ async fn coordinator_stop_is_prompt() { UNREACHABLE_CHAIN.to_string(), provider_account(), chain_state, + None, ) .start(); @@ -134,6 +145,7 @@ async fn coordinator_keeps_retrying_without_panicking() { UNREACHABLE_CHAIN.to_string(), provider_account(), chain_state.clone(), + None, ) .start(); @@ -155,6 +167,7 @@ async fn coordinator_releases_shared_state_after_stop() { UNREACHABLE_CHAIN.to_string(), provider_account(), chain_state.clone(), + None, ) .start(); @@ -577,6 +590,90 @@ async fn empty_block_does_not_refresh() { assert!(cs.provider_info.read().is_none()); } +// ── membership cache invalidation (invalidate_membership_on_events) ─────────── + +/// Resolver that counts calls, so tests can tell whether a lookup hit the cache or +/// went to the resolver. +struct CountingMembershipResolver { + calls: Arc, +} + +#[async_trait] +impl MembershipResolver for CountingMembershipResolver { + async fn fetch_members(&self, _bucket_id: u64) -> Result, String> { + self.calls.fetch_add(1, Ordering::SeqCst); + Ok(vec![]) + } +} + +fn member_removed_event(bucket_id: u64) -> StorageEvent { + StorageEvent::MemberRemoved { + bucket_id, + member: AccountId32::new([8u8; 32]), + block_hash: H256::zero(), + block_number: 1, + } +} + +#[tokio::test] +async fn membership_event_in_finalized_block_invalidates_cached_bucket() { + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let cache = MembershipCache::new( + Box::new(CountingMembershipResolver { + calls: calls.clone(), + }), + Duration::from_secs(300), + ); + let account = provider_account(); + + // Populate the cache for bucket 1. + cache.get_role(1, &account).await.unwrap(); + assert_eq!(calls.load(Ordering::SeqCst), 1); + + // A finalized block carrying a MemberRemoved event for bucket 1, alongside an + // unrelated provider event — mirroring what the coordinator parses per block. + let block_events = [ + registered_event(provider_account_2()), + member_removed_event(1), + ]; + invalidate_membership_on_events(&cache, &block_events); + + // The cached role can no longer be trusted — the next lookup must re-fetch. + cache.get_role(1, &account).await.unwrap(); + assert_eq!( + calls.load(Ordering::SeqCst), + 2, + "a membership event for the cached bucket must force a re-fetch" + ); +} + +#[tokio::test] +async fn unrelated_block_event_leaves_cached_membership_intact() { + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let cache = MembershipCache::new( + Box::new(CountingMembershipResolver { + calls: calls.clone(), + }), + Duration::from_secs(300), + ); + let account = provider_account(); + + cache.get_role(1, &account).await.unwrap(); + assert_eq!(calls.load(Ordering::SeqCst), 1); + + // A block with only a provider-lifecycle event (no bucket_id) must not touch + // a cached bucket's membership. + let block_events = [registered_event(provider_account())]; + invalidate_membership_on_events(&cache, &block_events); + + cache.get_role(1, &account).await.unwrap(); + assert_eq!( + calls.load(Ordering::SeqCst), + 1, + "an unrelated event must leave the cached bucket intact" + ); +} + // ─── Nonce counter persistence (nonce_store seeding) ────────────────────────── /// A minimal in-memory NonceStore for testing: holds the persisted value in a