diff --git a/CHANGELOG.md b/CHANGELOG.md index d491fe5781..365320a01d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,9 @@ All notable changes to this project will be documented in this file. ### Changes +- Serviceability + - New `SubscribeFeed` and `UnsubscribeFeed` instructions join or leave whole feeds on an EdgeSeat access pass in a single atomic transaction, charging one seat per feed rather than per group. A feed is all-or-nothing: the caller names feeds, the processor derives which groups change and rejects a group list that does not match, so two feeds carrying the same group stay unambiguous. `UpdateMulticastGroupRoles` now enforces the multicast-group allowlists for every access-pass type, EdgeSeat included, so purchased groups go through the feed instructions and individually comped groups through the allowlist. `CreateSubscribeUser` skips the allowlist only for the case its feed gate actually covers, closing two paths that could join a group with no check. `MAX_FEED_GROUPS` drops from 64 to 20, bounded by what one transaction can carry: because joining passes every group a feed holds, a larger feed could never be joined. No feed onchain is affected. For the same reason a user may hold at most 6 feeds, and a held feed the pass no longer carries is pruned on leave instead of blocking it. New errors: `EdgeSeatRequired` (101), `UserDeviceMismatch` (102), `UserFeedLimitExceeded` (103), `EdgeSeatIsMulticastOnly` (104). (#4109) + ## [v0.32.0](https://github.com/malbeclabs/doublezero/compare/client/v0.31.0...client/v0.32.0) - 2026-07-29 ### Breaking diff --git a/crates/doublezero-serviceability-instruction/src/multicastgroup.rs b/crates/doublezero-serviceability-instruction/src/multicastgroup.rs index dbf0267a84..d0a61caef3 100644 --- a/crates/doublezero-serviceability-instruction/src/multicastgroup.rs +++ b/crates/doublezero-serviceability-instruction/src/multicastgroup.rs @@ -21,7 +21,9 @@ use doublezero_serviceability::{ delete::MulticastGroupDeleteArgs, reactivate::MulticastGroupReactivateArgs, subscribe::UpdateMulticastGroupRolesArgs, + subscribe_feed::SubscribeFeedArgs, suspend::MulticastGroupSuspendArgs, + unsubscribe_feed::UnsubscribeFeedArgs, update::MulticastGroupUpdateArgs, }, resource::ResourceType, @@ -198,6 +200,91 @@ pub fn update_multicast_group_roles( ) } +/// `SubscribeFeed` (variant 117) — join whole feeds on an EdgeSeat access pass. +/// +/// Accounts: `[accesspass, user, globalstate, device, feeds.., groups..]`. +/// +/// `groups` must be exactly the groups this call adds; the processor derives that set from the feeds +/// and rejects a mismatch, so a stale client cannot half-apply a change. `feed_count` is derived from +/// `feeds.len()` rather than trusted from the caller. +pub fn subscribe_feed( + program_id: &Pubkey, + payer: &Pubkey, + accesspass: &Pubkey, + user: &Pubkey, + device: &Pubkey, + feeds: &[Pubkey], + groups: &[Pubkey], +) -> Instruction { + let (globalstate, _) = get_globalstate_pda(program_id); + let mut accounts = vec![ + AccountMeta::new(*accesspass, false), + AccountMeta::new(*user, false), + AccountMeta::new(globalstate, false), + AccountMeta::new_readonly(*device, false), + ]; + accounts.extend( + feeds + .iter() + .map(|feed| AccountMeta::new_readonly(*feed, false)), + ); + accounts.extend(groups.iter().map(|group| AccountMeta::new(*group, false))); + + common::build_with_permission( + program_id, + DoubleZeroInstruction::SubscribeFeed(SubscribeFeedArgs { + feed_count: feeds.len() as u8, + }), + accounts, + payer, + ) +} + +/// `UnsubscribeFeed` (variant 118) — leave whole feeds on an EdgeSeat access pass. +/// +/// Accounts: `[accesspass, user, globalstate, device, targets.., retained.., groups..]`. +/// +/// `retained` must be every feed the user keeps: two feeds on one pass can carry the same group, and +/// without the retained group sets the processor would drop a group another held feed still covers and +/// strand that feed's seat. Together `targets` and `retained` must cover every held feed still +/// provisioned on the pass; a held feed the pass dropped is pruned by the processor instead. +#[allow(clippy::too_many_arguments)] +pub fn unsubscribe_feed( + program_id: &Pubkey, + payer: &Pubkey, + accesspass: &Pubkey, + user: &Pubkey, + device: &Pubkey, + targets: &[Pubkey], + retained: &[Pubkey], + groups: &[Pubkey], +) -> Instruction { + let (globalstate, _) = get_globalstate_pda(program_id); + let mut accounts = vec![ + AccountMeta::new(*accesspass, false), + AccountMeta::new(*user, false), + AccountMeta::new(globalstate, false), + AccountMeta::new_readonly(*device, false), + ]; + accounts.extend( + targets + .iter() + .chain(retained) + .map(|feed| AccountMeta::new_readonly(*feed, false)), + ); + accounts.extend(groups.iter().map(|group| AccountMeta::new(*group, false))); + + common::build_with_permission( + program_id, + DoubleZeroInstruction::UnsubscribeFeed(UnsubscribeFeedArgs { + feed_count: targets.len() as u8, + retained_feed_count: retained.len() as u8, + }), + accounts, + payer, + ) +} + /// `AddMulticastGroupPubAllowlist` (variant 54). /// Accounts: `[mgroup, accesspass, globalstate, user_payer]`. pub fn add_multicast_group_pub_allowlist( diff --git a/smartcontract/programs/doublezero-serviceability/src/entrypoint.rs b/smartcontract/programs/doublezero-serviceability/src/entrypoint.rs index 46d357467b..c5dbfc6c5b 100644 --- a/smartcontract/programs/doublezero-serviceability/src/entrypoint.rs +++ b/smartcontract/programs/doublezero-serviceability/src/entrypoint.rs @@ -72,7 +72,9 @@ use crate::{ delete::process_delete_multicastgroup, reactivate::process_reactivate_multicastgroup, subscribe::process_update_multicastgroup_roles, + subscribe_feed::process_subscribe_feed, suspend::process_suspend_multicastgroup, + unsubscribe_feed::process_unsubscribe_feed, update::process_update_multicastgroup, }, permission::{ @@ -427,6 +429,12 @@ pub fn process_instruction( DoubleZeroInstruction::SetAccessPassFlags(value) => { process_set_access_pass_flags(program_id, accounts, &value)? } + DoubleZeroInstruction::SubscribeFeed(value) => { + process_subscribe_feed(program_id, accounts, &value)? + } + DoubleZeroInstruction::UnsubscribeFeed(value) => { + process_unsubscribe_feed(program_id, accounts, &value)? + } }; Ok(()) } diff --git a/smartcontract/programs/doublezero-serviceability/src/error.rs b/smartcontract/programs/doublezero-serviceability/src/error.rs index c1fd9c46bc..74d9564ba0 100644 --- a/smartcontract/programs/doublezero-serviceability/src/error.rs +++ b/smartcontract/programs/doublezero-serviceability/src/error.rs @@ -208,6 +208,14 @@ pub enum DoubleZeroError { "Feed billing window is invalid (window_end must be in the future and <= terminates_at)" )] FeedInvalidBillingWindow, // variant 100 + #[error("Feed subscriptions require an EdgeSeat access pass")] + EdgeSeatRequired, // variant 101 + #[error("The passed Device account is not the user's device")] + UserDeviceMismatch, // variant 102 + #[error("User would hold more feeds than an access pass may carry")] + UserFeedLimitExceeded, // variant 103 + #[error("An EdgeSeat feed seat is only held by a Multicast user")] + EdgeSeatIsMulticastOnly, // variant 104 } impl From for ProgramError { @@ -314,6 +322,10 @@ impl From for ProgramError { DoubleZeroError::FeedMaxFutureUsersBelowMaxUsers => ProgramError::Custom(98), DoubleZeroError::FeedInvalidAnniversaryDay => ProgramError::Custom(99), DoubleZeroError::FeedInvalidBillingWindow => ProgramError::Custom(100), + DoubleZeroError::EdgeSeatRequired => ProgramError::Custom(101), + DoubleZeroError::UserDeviceMismatch => ProgramError::Custom(102), + DoubleZeroError::UserFeedLimitExceeded => ProgramError::Custom(103), + DoubleZeroError::EdgeSeatIsMulticastOnly => ProgramError::Custom(104), } } } @@ -421,6 +433,10 @@ impl From for DoubleZeroError { 98 => DoubleZeroError::FeedMaxFutureUsersBelowMaxUsers, 99 => DoubleZeroError::FeedInvalidAnniversaryDay, 100 => DoubleZeroError::FeedInvalidBillingWindow, + 101 => DoubleZeroError::EdgeSeatRequired, + 102 => DoubleZeroError::UserDeviceMismatch, + 103 => DoubleZeroError::UserFeedLimitExceeded, + 104 => DoubleZeroError::EdgeSeatIsMulticastOnly, _ => DoubleZeroError::Custom(e), } } diff --git a/smartcontract/programs/doublezero-serviceability/src/instructions.rs b/smartcontract/programs/doublezero-serviceability/src/instructions.rs index 19efa81f3f..13b60940e9 100644 --- a/smartcontract/programs/doublezero-serviceability/src/instructions.rs +++ b/smartcontract/programs/doublezero-serviceability/src/instructions.rs @@ -58,7 +58,9 @@ use crate::processors::{ delete::MulticastGroupDeleteArgs, reactivate::MulticastGroupReactivateArgs, subscribe::UpdateMulticastGroupRolesArgs, + subscribe_feed::SubscribeFeedArgs, suspend::MulticastGroupSuspendArgs, + unsubscribe_feed::UnsubscribeFeedArgs, update::MulticastGroupUpdateArgs, }, permission::{ @@ -253,6 +255,9 @@ pub enum DoubleZeroInstruction { DeleteFeed(FeedDeleteArgs), // variant 114 SetAccessPassFeeds(SetAccessPassFeedsArgs), // variant 115 SetAccessPassFlags(SetAccessPassFlagsArgs), // variant 116 + + SubscribeFeed(SubscribeFeedArgs), // variant 117 + UnsubscribeFeed(UnsubscribeFeedArgs), // variant 118 } impl DoubleZeroInstruction { @@ -401,6 +406,9 @@ impl DoubleZeroInstruction { 115 => Ok(Self::SetAccessPassFeeds(SetAccessPassFeedsArgs::try_from(rest).unwrap())), 116 => Ok(Self::SetAccessPassFlags(SetAccessPassFlagsArgs::try_from(rest).unwrap())), + 117 => Ok(Self::SubscribeFeed(SubscribeFeedArgs::try_from(rest).unwrap())), + 118 => Ok(Self::UnsubscribeFeed(UnsubscribeFeedArgs::try_from(rest).unwrap())), + _ => Err(ProgramError::InvalidInstructionData), } } @@ -550,6 +558,8 @@ impl DoubleZeroInstruction { Self::DeleteFeed(_) => "DeleteFeed".to_string(), // variant 114 Self::SetAccessPassFeeds(_) => "SetAccessPassFeeds".to_string(), // variant 115 Self::SetAccessPassFlags(_) => "SetAccessPassFlags".to_string(), // variant 116 + Self::SubscribeFeed(_) => "SubscribeFeed".to_string(), // variant 117 + Self::UnsubscribeFeed(_) => "UnsubscribeFeed".to_string(), // variant 118 } } @@ -692,6 +702,8 @@ impl DoubleZeroInstruction { Self::DeleteFeed(args) => format!("{args:?}"), // variant 114 Self::SetAccessPassFeeds(args) => format!("{args:?}"), // variant 115 Self::SetAccessPassFlags(args) => format!("{args:?}"), // variant 116 + Self::SubscribeFeed(args) => format!("{args:?}"), // variant 117 + Self::UnsubscribeFeed(args) => format!("{args:?}"), // variant 118 } } } diff --git a/smartcontract/programs/doublezero-serviceability/src/processors/feed/create.rs b/smartcontract/programs/doublezero-serviceability/src/processors/feed/create.rs index ad6fb66b59..65e21eefcc 100644 --- a/smartcontract/programs/doublezero-serviceability/src/processors/feed/create.rs +++ b/smartcontract/programs/doublezero-serviceability/src/processors/feed/create.rs @@ -22,8 +22,9 @@ use solana_program::{ /// Maximum `name` length, matching the Exchange/Location `name` cap. pub const MAX_FEED_NAME_LEN: usize = 64; -/// Maximum number of multicast groups in a feed. -pub const MAX_FEED_GROUPS: usize = 64; +/// Maximum number of multicast groups in a feed. A feed's whole group set joins in one +/// `SubscribeFeed` transaction, so this is bounded by transaction capacity. +pub const MAX_FEED_GROUPS: usize = 20; #[derive(BorshSerialize, BorshDeserializeIncremental, PartialEq, Debug, Clone, Default)] pub struct FeedCreateArgs { diff --git a/smartcontract/programs/doublezero-serviceability/src/processors/multicastgroup/feed.rs b/smartcontract/programs/doublezero-serviceability/src/processors/multicastgroup/feed.rs new file mode 100644 index 0000000000..c40e0e0e10 --- /dev/null +++ b/smartcontract/programs/doublezero-serviceability/src/processors/multicastgroup/feed.rs @@ -0,0 +1,280 @@ +//! Shared setup for [`super::subscribe_feed`] and [`super::unsubscribe_feed`]. +//! +//! A feed is all-or-nothing: the user's subscribed groups are the union of the groups carried by the +//! feeds they hold. Both instructions name feeds and derive the groups that change, so a seat follows +//! the feed the caller named rather than being inferred from group membership. That is what makes two +//! feeds carrying the same group unambiguous. + +use crate::{ + authorize::{authorize, split_trailing_permission}, + error::DoubleZeroError, + pda::{get_accesspass_pda, get_globalstate_pda}, + processors::{feed::check_feed_metro_coverage, validation::validate_program_account}, + serializer::try_acc_write, + state::{ + accesspass::{AccessPass, AccessPassType}, + device::Device, + feed::Feed, + globalstate::GlobalState, + user::{User, UserType}, + }, +}; +use solana_program::{ + account_info::{next_account_info, AccountInfo}, + entrypoint::ProgramResult, + msg, + program_error::ProgramError, + pubkey::Pubkey, +}; +use std::net::Ipv4Addr; + +use super::subscribe::update_user_multicastgroup_roles; + +/// The validated accounts and state both feed instructions work from. +pub struct FeedSubscriptionContext<'a, 'b> { + pub accesspass_account: &'a AccountInfo<'b>, + pub user_account: &'a AccountInfo<'b>, + pub payer_account: &'a AccountInfo<'b>, + pub permission_account: Option<&'a AccountInfo<'b>>, + pub globalstate: GlobalState, + pub accesspass: AccessPass, + pub user: User, + pub device: Device, + /// The variable account section: feeds followed by multicast groups. + pub variable: Vec<&'a AccountInfo<'b>>, +} + +/// Parse and validate `[accesspass, user, globalstate, device, ..variable.., payer, system, perm?]`. +/// +/// Applies every check both instructions share: the pass exists and is EdgeSeat, the user is +/// Multicast, and the device is the user's own. +pub fn load_context<'a, 'b>( + program_id: &Pubkey, + accounts: &'a [AccountInfo<'b>], +) -> Result, ProgramError> { + let accounts_iter = &mut accounts.iter(); + + let accesspass_account = next_account_info(accounts_iter)?; + let user_account = next_account_info(accounts_iter)?; + let gs_account = next_account_info(accounts_iter)?; + let device_account = next_account_info(accounts_iter)?; + + let remaining: Vec<&AccountInfo> = accounts_iter.collect(); + let (payer_account, system_program, variable, permission_account) = + split_trailing_permission(program_id, &remaining)?; + + assert!(payer_account.is_signer, "Payer must be a signer"); + assert_eq!( + *system_program.unsigned_key(), + solana_system_interface::program::ID, + "Invalid System Program Account Owner" + ); + + let (expected_globalstate_pda, _) = get_globalstate_pda(program_id); + assert_eq!( + gs_account.key, &expected_globalstate_pda, + "Invalid GlobalState PDA" + ); + let globalstate = GlobalState::try_from(gs_account)?; + + if accesspass_account.data_is_empty() { + return Err(DoubleZeroError::AccessPassNotFound.into()); + } + validate_program_account!( + accesspass_account, + program_id, + writable = true, + "AccessPass" + ); + validate_program_account!(user_account, program_id, writable = true, "User"); + validate_program_account!(device_account, program_id, writable = false, "Device"); + + let user = User::try_from(user_account)?; + if user.user_type != UserType::Multicast { + msg!( + "a feed subscription requires a Multicast user, got {}", + user.user_type + ); + return Err(DoubleZeroError::EdgeSeatIsMulticastOnly.into()); + } + if device_account.key != &user.device_pk { + msg!( + "device {} is not the user's device {}", + device_account.key, + user.device_pk + ); + return Err(DoubleZeroError::UserDeviceMismatch.into()); + } + let device = Device::try_from(device_account)?; + + let accesspass = AccessPass::try_from(accesspass_account)?; + let (accesspass_pda, _) = get_accesspass_pda(program_id, &user.client_ip, &user.owner); + let (accesspass_dynamic_pda, _) = + get_accesspass_pda(program_id, &Ipv4Addr::UNSPECIFIED, &user.owner); + assert!( + accesspass_account.key == &accesspass_pda + || accesspass_account.key == &accesspass_dynamic_pda, + "Invalid AccessPass PDA", + ); + if !matches!(accesspass.accesspass_type, AccessPassType::EdgeSeat(_)) { + msg!( + "AccessPass type {:?} carries no feeds; use UpdateMulticastGroupRoles", + accesspass.accesspass_type + ); + return Err(DoubleZeroError::EdgeSeatRequired.into()); + } + + Ok(FeedSubscriptionContext { + accesspass_account, + user_account, + payer_account, + permission_account, + globalstate, + accesspass, + user, + device, + variable: variable.to_vec(), + }) +} + +impl FeedSubscriptionContext<'_, '_> { + /// The pass's own `user_payer`, a foundation member, or a holder of `required_flag`. + pub fn authorize_payer( + &self, + program_id: &Pubkey, + required_flag: u128, + denied: DoubleZeroError, + ) -> ProgramResult { + if self.accesspass.user_payer == *self.payer_account.key + || self + .globalstate + .foundation_allowlist + .contains(self.payer_account.key) + { + return Ok(()); + } + if authorize( + program_id, + &mut self.permission_account.into_iter(), + self.payer_account.key, + &self.globalstate, + required_flag, + ) + .is_err() + { + msg!( + "AccessPass user_payer {:?} does not match payer {:?}", + self.accesspass.user_payer, + self.payer_account.key + ); + return Err(denied.into()); + } + Ok(()) + } +} + +/// Read a run of Feed accounts, checking each is on the pass and serves the user's metro. +/// +/// A feed listed in `held` is loaded even if no longer provisioned on the pass (its seat is gone, +/// so there is no metro to validate against); strict callers pass `&[]`. +pub fn load_feeds( + program_id: &Pubkey, + accesspass: &AccessPass, + device_exchange: &Pubkey, + feed_accounts: &[&AccountInfo], + held: &[Pubkey], +) -> Result, ProgramError> { + let mut feeds: Vec<(Pubkey, Feed)> = Vec::with_capacity(feed_accounts.len()); + for feed_account in feed_accounts { + let provisioned = accesspass + .feed_seats() + .iter() + .any(|s| s.feed_key == *feed_account.key); + if provisioned { + check_feed_metro_coverage( + program_id, + accesspass, + device_exchange, + None, + Some(feed_account), + )?; + } else if held.contains(feed_account.key) { + if feed_account.owner != program_id { + return Err(DoubleZeroError::InvalidAccountOwner.into()); + } + } else { + msg!( + "Feed {} is not provisioned on the access pass", + feed_account.key + ); + return Err(DoubleZeroError::FeedNotOnAccessPass.into()); + } + if feeds.iter().any(|(key, _)| key == feed_account.key) { + msg!("feed {} passed more than once", feed_account.key); + return Err(DoubleZeroError::InvalidArgument.into()); + } + feeds.push((*feed_account.key, Feed::try_from(*feed_account)?)); + } + Ok(feeds) +} + +/// Reject unless `group_accounts` is exactly `expected`, so a stale client cannot half-apply a change. +pub fn check_group_accounts(group_accounts: &[&AccountInfo], expected: &[Pubkey]) -> ProgramResult { + if group_accounts.len() != expected.len() { + msg!( + "passed {} MulticastGroup accounts but this call changes {}", + group_accounts.len(), + expected.len() + ); + return Err(DoubleZeroError::InvalidArgument.into()); + } + if let Some(missing) = expected + .iter() + .find(|group| !group_accounts.iter().any(|a| a.key == *group)) + { + msg!("group {} changes here but was not passed", missing); + return Err(DoubleZeroError::InvalidArgument.into()); + } + Ok(()) +} + +/// Apply `subscriber` to every passed group and write each group account back. +/// +/// The publisher role is carried through unchanged: clearing it would deallocate the user's `dz_ip` +/// as a side effect. +pub fn apply_groups( + program_id: &Pubkey, + accounts: &[AccountInfo], + group_accounts: &[&AccountInfo], + user: &mut User, + payer_account: &AccountInfo, + subscriber: bool, +) -> ProgramResult { + for group_account in group_accounts { + validate_program_account!( + *group_account, + program_id, + writable = true, + "MulticastGroup" + ); + let carry_publisher = user.publishers.contains(group_account.key); + let result = + update_user_multicastgroup_roles(group_account, user, carry_publisher, subscriber)?; + try_acc_write(&result.mgroup, group_account, payer_account, accounts)?; + } + Ok(()) +} + +/// Persist the pass and user after a seat change. +pub fn write_back( + ctx_accesspass: &AccessPass, + ctx_user: &User, + accesspass_account: &AccountInfo, + user_account: &AccountInfo, + payer_account: &AccountInfo, + accounts: &[AccountInfo], +) -> ProgramResult { + try_acc_write(ctx_accesspass, accesspass_account, payer_account, accounts)?; + try_acc_write(ctx_user, user_account, payer_account, accounts)?; + Ok(()) +} diff --git a/smartcontract/programs/doublezero-serviceability/src/processors/multicastgroup/mod.rs b/smartcontract/programs/doublezero-serviceability/src/processors/multicastgroup/mod.rs index b79c403b5d..aec86cccea 100644 --- a/smartcontract/programs/doublezero-serviceability/src/processors/multicastgroup/mod.rs +++ b/smartcontract/programs/doublezero-serviceability/src/processors/multicastgroup/mod.rs @@ -1,7 +1,10 @@ pub mod allowlist; pub mod create; pub mod delete; +pub mod feed; pub mod reactivate; pub mod subscribe; +pub mod subscribe_feed; pub mod suspend; +pub mod unsubscribe_feed; pub mod update; diff --git a/smartcontract/programs/doublezero-serviceability/src/processors/multicastgroup/subscribe.rs b/smartcontract/programs/doublezero-serviceability/src/processors/multicastgroup/subscribe.rs index c053131245..0c0f09e9d0 100644 --- a/smartcontract/programs/doublezero-serviceability/src/processors/multicastgroup/subscribe.rs +++ b/smartcontract/programs/doublezero-serviceability/src/processors/multicastgroup/subscribe.rs @@ -3,15 +3,13 @@ use crate::{ error::DoubleZeroError, pda::{get_accesspass_pda, get_globalstate_pda, get_resource_extension_pda}, processors::{ - feed::check_feed_metro_coverage, resource::{allocate_ip, deallocate_ip}, validation::validate_program_account, }, resource::ResourceType, serializer::try_acc_write, state::{ - accesspass::{AccessPass, AccessPassType}, - device::Device, + accesspass::AccessPass, globalstate::GlobalState, multicastgroup::{MulticastGroup, MulticastGroupStatus}, permission::permission_flags, @@ -52,20 +50,48 @@ impl fmt::Debug for UpdateMulticastGroupRolesArgs { pub struct SubscribeUserResult { pub mgroup: MulticastGroup, /// True if the publisher list transitioned between empty and non-empty - /// (gained first publisher or lost last publisher). Callers that need to - /// trigger downstream reprocessing should check this flag. + /// (gained first publisher or lost last publisher). pub publisher_list_transitioned: bool, } -/// Toggle a user's multicast group roles. +/// Authorize a role grant against the access pass's multicast-group allowlists. +/// +/// This is the authorization used by every access-pass type that grants groups individually. The +/// only alternative is the EdgeSeat feed metro gate, which derives joinable groups from the feeds +/// provisioned on the pass; a caller runs one or the other, never neither. Removals are always +/// allowed, so a user can be cleaned up after a group leaves an allowlist. +pub fn check_mgroup_allowlists( + accesspass: &AccessPass, + mgroup_key: &Pubkey, + publisher: bool, + subscriber: bool, +) -> ProgramResult { + if publisher && !accesspass.mgroup_pub_allowlist.contains(mgroup_key) { + msg!("{:?}", accesspass); + return Err(DoubleZeroError::NotAllowed.into()); + } + if subscriber && !accesspass.mgroup_sub_allowlist.contains(mgroup_key) { + msg!("{:?}", accesspass); + return Err(DoubleZeroError::NotAllowed.into()); + } + Ok(()) +} + +/// Set a user's multicast group roles to the requested state. +/// +/// `publisher` and `subscriber` are desired states, not toggle signals: `true` adds the role when the +/// user does not hold it, `false` removes it when they do, and either is a no-op otherwise. +/// +/// Mechanics only: this does NOT authorize the change. Callers must first run either +/// [`check_mgroup_allowlists`] or the EdgeSeat feed metro gate, so that the authorization a +/// processor performed is visible in that processor rather than claimed through an argument here. /// /// Handles both create-time subscription (user lists start empty, only adds) -/// and post-activation subscription changes (add/remove toggle). The caller is +/// and post-activation subscription changes. The caller is /// responsible for setting `user.status = Updating` when /// `publisher_list_transitioned` is true and the user is already activated. pub fn update_user_multicastgroup_roles( mgroup_account: &AccountInfo, - accesspass: &AccessPass, user: &mut User, publisher: bool, subscriber: bool, @@ -76,20 +102,6 @@ pub fn update_user_multicastgroup_roles( return Err(DoubleZeroError::InvalidStatus.into()); } - // Check allowlists for additions. EdgeSeat passes derive joinable groups from their feeds' - // metro→group map (the feed metro gate), which supersedes the mgroup allowlist; the caller is - // responsible for running enforce_feed_metro_gate for EdgeSeat connects. - let is_edge_seat = matches!(accesspass.accesspass_type, AccessPassType::EdgeSeat(_)); - if publisher && !is_edge_seat && !accesspass.mgroup_pub_allowlist.contains(mgroup_account.key) { - msg!("{:?}", accesspass); - return Err(DoubleZeroError::NotAllowed.into()); - } - if subscriber && !is_edge_seat && !accesspass.mgroup_sub_allowlist.contains(mgroup_account.key) - { - msg!("{:?}", accesspass); - return Err(DoubleZeroError::NotAllowed.into()); - } - let mut publisher_list_transitioned = false; // Manage the publisher list @@ -162,18 +174,12 @@ pub fn process_update_multicastgroup_roles( let globalstate = GlobalState::try_from(gs_account)?; let multicast_publisher_block_ext = next_account_info(accounts_iter)?; - // Trailing layout: [device?, feed?, payer, system, permission?]. The SDK appends the payer's - // Permission PDA last (via execute_authorized_transaction); the optional EdgeSeat device/feed - // accounts for post-activation metro re-gating precede payer/system, because the client pushes - // them into the instruction's account list ahead of the [payer, system, permission] trailer - // that assemble_instructions always appends. split_trailing_permission identifies the - // Permission by PDA match rather than by position, so it never mistakes device/feed for the - // Permission account regardless of which optional accounts are present. + // Trailing layout: [payer, system, permission?]. The SDK appends the payer's Permission PDA last + // (via execute_authorized_transaction), and split_trailing_permission identifies it by PDA match + // rather than by position. let remaining: Vec<&AccountInfo> = accounts_iter.collect(); - let (payer_account, system_program, leading, permission_account) = + let (payer_account, system_program, _leading, permission_account) = split_trailing_permission(program_id, &remaining)?; - let device_account = leading.first().copied(); - let feed_account = leading.get(1).copied(); #[cfg(test)] msg!("process_update_multicastgroup_roles({:?})", value); @@ -229,8 +235,7 @@ pub fn process_update_multicastgroup_roles( // must be a foundation member, or — for removal-only cleanup (no roles being // granted) — hold USER_ADMIN. The USER_ADMIN path lets an operator strip a // user's multicast roles as a prerequisite to deleting/request-banning that - // user (see DeleteUserCommand / RequestBanUserCommand, which authorize the - // final instruction with the same USER_ADMIN flag). + // user. if accesspass.user_payer != *payer_account.key && !globalstate.foundation_allowlist.contains(payer_account.key) { @@ -279,30 +284,16 @@ pub fn process_update_multicastgroup_roles( } } - // EdgeSeat passes derive joinable groups from their feeds' metro→group map. The seat was - // ticked at connect (CreateSubscribeUser), so post-activation role adds only re-validate - // coverage; they do not re-tick. - if matches!(accesspass.accesspass_type, AccessPassType::EdgeSeat(_)) - && (value.publisher || value.subscriber) - { - let device_account = device_account.ok_or(DoubleZeroError::MetroMismatch)?; - validate_program_account!(device_account, program_id, writable = false, "Device"); - if user.device_pk != *device_account.key { - return Err(ProgramError::InvalidAccountData); - } - let device = Device::try_from(device_account)?; - check_feed_metro_coverage( - program_id, - &accesspass, - &device.exchange_pk, - Some(mgroup_account.key), - feed_account, - )?; - } + // Every pass type is authorized the same way here: the group must be on the pass's allowlist. + check_mgroup_allowlists( + &accesspass, + mgroup_account.key, + value.publisher, + value.subscriber, + )?; let result = update_user_multicastgroup_roles( mgroup_account, - &accesspass, &mut user, value.publisher, value.subscriber, diff --git a/smartcontract/programs/doublezero-serviceability/src/processors/multicastgroup/subscribe_feed.rs b/smartcontract/programs/doublezero-serviceability/src/processors/multicastgroup/subscribe_feed.rs new file mode 100644 index 0000000000..f40007c54d --- /dev/null +++ b/smartcontract/programs/doublezero-serviceability/src/processors/multicastgroup/subscribe_feed.rs @@ -0,0 +1,118 @@ +//! `SubscribeFeed` (variant 117) — join whole feeds on an EdgeSeat access pass. +//! +//! Accounts: `[accesspass, user, globalstate, device, feeds.., groups.., payer, system, perm?]`. +//! One seat is charged per feed newly held. See [`super::feed`] for the shared model. + +use crate::{ + error::DoubleZeroError, + processors::multicastgroup::feed::{ + apply_groups, check_group_accounts, load_context, load_feeds, write_back, + }, + state::{permission::permission_flags, user::UserStatus}, +}; +use borsh::BorshSerialize; +use borsh_incremental::BorshDeserializeIncremental; +use solana_program::{account_info::AccountInfo, entrypoint::ProgramResult, msg, pubkey::Pubkey}; +use std::fmt; + +/// Cap on the feeds one user may hold at once. Bounded by what a single [`UnsubscribeFeed`] +/// transaction can name. +pub const MAX_USER_FEEDS: usize = 6; + +#[derive(BorshSerialize, BorshDeserializeIncremental, PartialEq, Clone)] +pub struct SubscribeFeedArgs { + /// How many of the variable accounts are Feeds. The rest are the MulticastGroups being joined. + pub feed_count: u8, +} + +impl fmt::Debug for SubscribeFeedArgs { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "feed_count: {:?}", self.feed_count) + } +} + +pub fn process_subscribe_feed( + program_id: &Pubkey, + accounts: &[AccountInfo], + value: &SubscribeFeedArgs, +) -> ProgramResult { + let mut ctx = load_context(program_id, accounts)?; + + #[cfg(test)] + msg!("process_subscribe_feed({:?})", value); + + if ctx.user.status != UserStatus::Activated { + msg!("UserStatus: {:?}", ctx.user.status); + return Err(DoubleZeroError::InvalidStatus.into()); + } + + // Joining consumes the pass's paid capacity. + ctx.authorize_payer( + program_id, + permission_flags::ACCESS_PASS_ADMIN, + DoubleZeroError::Unauthorized, + )?; + + let feed_count = value.feed_count as usize; + if feed_count == 0 || ctx.variable.len() < feed_count { + msg!( + "bad account shape: feed_count={}, variable={}", + feed_count, + ctx.variable.len() + ); + return Err(DoubleZeroError::InvalidArgument.into()); + } + let (feed_accounts, group_accounts) = ctx.variable.split_at(feed_count); + + let feeds = load_feeds( + program_id, + &ctx.accesspass, + &ctx.device.exchange_pk, + feed_accounts, + &[], + )?; + + let mut expected: Vec = Vec::new(); + for (_, feed) in &feeds { + for group in &feed.groups { + if !ctx.user.subscribers.contains(group) && !expected.contains(group) { + expected.push(*group); + } + } + } + check_group_accounts(group_accounts, &expected)?; + + apply_groups( + program_id, + accounts, + group_accounts, + &mut ctx.user, + ctx.payer_account, + true, + )?; + + for (feed_key, _) in &feeds { + if ctx.user.feed_pks.contains(feed_key) { + continue; + } + if ctx.user.feed_pks.len() >= MAX_USER_FEEDS { + msg!( + "user already holds {} feeds; a user may hold at most {}", + ctx.user.feed_pks.len(), + MAX_USER_FEEDS + ); + return Err(DoubleZeroError::UserFeedLimitExceeded.into()); + } + ctx.accesspass.try_add_feed_user(feed_key)?; + ctx.user.feed_pks.push(*feed_key); + } + + write_back( + &ctx.accesspass, + &ctx.user, + ctx.accesspass_account, + ctx.user_account, + ctx.payer_account, + accounts, + ) +} diff --git a/smartcontract/programs/doublezero-serviceability/src/processors/multicastgroup/unsubscribe_feed.rs b/smartcontract/programs/doublezero-serviceability/src/processors/multicastgroup/unsubscribe_feed.rs new file mode 100644 index 0000000000..ac5d570282 --- /dev/null +++ b/smartcontract/programs/doublezero-serviceability/src/processors/multicastgroup/unsubscribe_feed.rs @@ -0,0 +1,161 @@ +//! `UnsubscribeFeed` (variant 118) — leave whole feeds on an EdgeSeat access pass. +//! +//! Accounts: +//! `[accesspass, user, globalstate, device, targets.., retained.., groups.., payer, system, perm?]`. +//! +//! `retained` is every feed the user keeps. It is required, not optional: two feeds on one pass can +//! carry the same group, so without the retained feeds' group sets a departing feed would drop a +//! group another held feed still covers, and leave that feed's seat charged against a user holding +//! nothing in it. A held feed the pass no longer carries is exempt: it has no seat to strand, is +//! pruned on any leave, and may itself be a target to unsubscribe its groups (but cannot stand in +//! as retained). See [`super::feed`] for the shared model. + +use crate::{ + error::DoubleZeroError, + processors::multicastgroup::feed::{ + apply_groups, check_group_accounts, load_context, load_feeds, write_back, + }, + state::permission::permission_flags, +}; +use borsh::BorshSerialize; +use borsh_incremental::BorshDeserializeIncremental; +use solana_program::{account_info::AccountInfo, entrypoint::ProgramResult, msg, pubkey::Pubkey}; +use std::fmt; + +#[derive(BorshSerialize, BorshDeserializeIncremental, PartialEq, Clone)] +pub struct UnsubscribeFeedArgs { + /// Feeds being left. + pub feed_count: u8, + /// Feeds the user keeps. Together with `feed_count` this must cover every feed the user holds + /// that is still provisioned on the pass. + pub retained_feed_count: u8, +} + +impl fmt::Debug for UnsubscribeFeedArgs { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + f, + "feed_count: {:?}, retained_feed_count: {:?}", + self.feed_count, self.retained_feed_count + ) + } +} + +pub fn process_unsubscribe_feed( + program_id: &Pubkey, + accounts: &[AccountInfo], + value: &UnsubscribeFeedArgs, +) -> ProgramResult { + let mut ctx = load_context(program_id, accounts)?; + + #[cfg(test)] + msg!("process_unsubscribe_feed({:?})", value); + + // No status check: a user created but not yet activated must still be cleanable. + ctx.authorize_payer( + program_id, + permission_flags::USER_ADMIN, + DoubleZeroError::NotAllowed, + )?; + + let feed_count = value.feed_count as usize; + let retained_count = value.retained_feed_count as usize; + if feed_count == 0 || ctx.variable.len() < feed_count + retained_count { + msg!( + "bad account shape: feed_count={}, retained_feed_count={}, variable={}", + feed_count, + retained_count, + ctx.variable.len() + ); + return Err(DoubleZeroError::InvalidArgument.into()); + } + let (target_accounts, rest) = ctx.variable.split_at(feed_count); + let (retained_accounts, group_accounts) = rest.split_at(retained_count); + + let targets = load_feeds( + program_id, + &ctx.accesspass, + &ctx.device.exchange_pk, + target_accounts, + &ctx.user.feed_pks, + )?; + let retained = load_feeds( + program_id, + &ctx.accesspass, + &ctx.device.exchange_pk, + retained_accounts, + &[], + )?; + + // Without both checks a caller can release a seat while keeping the groups. + for (key, _) in &retained { + if targets.iter().any(|(target, _)| target == key) { + msg!("feed {} passed as both a target and retained", key); + return Err(DoubleZeroError::InvalidArgument.into()); + } + if !ctx.user.feed_pks.contains(key) { + msg!("retained feed {} is not held by this user", key); + return Err(DoubleZeroError::InvalidArgument.into()); + } + } + + // A held feed no longer on the pass has no seat to strand: prune it instead of requiring it. + let provisioned: Vec = ctx + .accesspass + .feed_seats() + .iter() + .map(|s| s.feed_key) + .collect(); + ctx.user.feed_pks.retain(|held| { + if provisioned.contains(held) { + return true; + } + msg!("held feed {} is no longer on the pass; pruning it", held); + false + }); + + for held in &ctx.user.feed_pks { + if !targets.iter().any(|(key, _)| key == held) + && !retained.iter().any(|(key, _)| key == held) + { + msg!("feed {} is held by this user and must be passed", held); + return Err(DoubleZeroError::FeedAccountRequired.into()); + } + } + + let mut expected: Vec = Vec::new(); + for (_, feed) in &targets { + for group in &feed.groups { + let still_covered = retained.iter().any(|(_, r)| r.groups.contains(group)); + if ctx.user.subscribers.contains(group) && !still_covered && !expected.contains(group) { + expected.push(*group); + } + } + } + check_group_accounts(group_accounts, &expected)?; + + apply_groups( + program_id, + accounts, + group_accounts, + &mut ctx.user, + ctx.payer_account, + false, + )?; + + for (feed_key, _) in &targets { + if ctx.user.feed_pks.contains(feed_key) { + ctx.accesspass.remove_feed_user(feed_key); + ctx.user.feed_pks.retain(|k| k != feed_key); + } + } + + write_back( + &ctx.accesspass, + &ctx.user, + ctx.accesspass_account, + ctx.user_account, + ctx.payer_account, + accounts, + ) +} diff --git a/smartcontract/programs/doublezero-serviceability/src/processors/user/create_subscribe.rs b/smartcontract/programs/doublezero-serviceability/src/processors/user/create_subscribe.rs index 326fd09a6d..222fb4ad64 100644 --- a/smartcontract/programs/doublezero-serviceability/src/processors/user/create_subscribe.rs +++ b/smartcontract/programs/doublezero-serviceability/src/processors/user/create_subscribe.rs @@ -20,7 +20,12 @@ use super::{ create_core::{create_user_core, CreateUserCoreAccounts, PDAVersion}, resource_onchain_helpers, }; -use crate::processors::multicastgroup::subscribe::update_user_multicastgroup_roles; +use crate::{ + processors::multicastgroup::subscribe::{ + check_mgroup_allowlists, update_user_multicastgroup_roles, + }, + state::accesspass::AccessPassType, +}; #[derive(BorshSerialize, BorshDeserializeIncremental, PartialEq, Clone)] pub struct UserCreateSubscribeArgs { @@ -135,10 +140,24 @@ pub fn process_create_subscribe_user( feed_account, )?; + // Mirrors the exact condition under which `create_user_core` ran the feed metro gate + // (create_core.rs). Keying the skip on pass type alone would let an EdgeSeat pass join a group + // with neither check: the gate is multicast-only and would not have run. The publisher allowlist + // is always checked, since a feed sells receive only and grants no publisher role. + let feed_gated = matches!( + result.accesspass.accesspass_type, + AccessPassType::EdgeSeat(_) + ) && value.user_type == UserType::Multicast; + check_mgroup_allowlists( + &result.accesspass, + mgroup_account.key, + value.publisher, + value.subscriber && !feed_gated, + )?; + // Subscribe user to multicast group let subscribe_result = update_user_multicastgroup_roles( mgroup_account, - &result.accesspass, &mut result.user, value.publisher, value.subscriber, diff --git a/smartcontract/programs/doublezero-serviceability/tests/feed_subscription_test.rs b/smartcontract/programs/doublezero-serviceability/tests/feed_subscription_test.rs new file mode 100644 index 0000000000..540908aa75 --- /dev/null +++ b/smartcontract/programs/doublezero-serviceability/tests/feed_subscription_test.rs @@ -0,0 +1,1053 @@ +//! `UpdateFeedSubscription` (variant 117) — joining and leaving whole feeds on an EdgeSeat pass. +//! +//! The property under test throughout is that a seat is held per *feed*, not per group: three +//! groups inside one feed cost one seat, a second feed costs a second, and a seat is released only +//! when the user's last group in that feed goes away. + +use doublezero_serviceability::{ + instructions::DoubleZeroInstruction, + pda::{ + get_accesspass_pda, get_device_pda, get_feed_pda, get_multicastgroup_pda, + get_resource_extension_pda, get_user_pda, + }, + processors::{ + accesspass::{ + set::SetAccessPassArgs, + set_feeds::{FeedSeatConfig, SetAccessPassFeedsArgs}, + }, + device::{create::DeviceCreateArgs, update::DeviceUpdateArgs}, + feed::create::{FeedCreateArgs, MAX_FEED_GROUPS}, + multicastgroup::{ + create::MulticastGroupCreateArgs, + subscribe::UpdateMulticastGroupRolesArgs, + subscribe_feed::{SubscribeFeedArgs, MAX_USER_FEEDS}, + unsubscribe_feed::UnsubscribeFeedArgs, + }, + user::{create::UserCreateArgs, create_subscribe::UserCreateSubscribeArgs}, + }, + resource::ResourceType, + state::{ + accesspass::{AccessPass, AccessPassType, FeedSeat}, + device::DeviceType, + user::{User, UserCYOA, UserType}, + }, +}; +use solana_program_test::*; +use solana_sdk::{ + instruction::{AccountMeta, InstructionError}, + pubkey::Pubkey, + signature::Signer, + transaction::TransactionError, +}; +use std::net::Ipv4Addr; + +mod test_helpers; +use test_helpers::*; + +// Far-future billing-window bounds so the "window_end must be in the future" check stays satisfied. +const TEST_WINDOW_END: i64 = 4_000_000_000; +const TEST_TERMINATES_AT: i64 = 4_100_000_000; + +struct Fixture { + banks_client: BanksClient, + payer: solana_sdk::signature::Keypair, + program_id: Pubkey, + globalstate_pubkey: Pubkey, + exchange_pubkey: Pubkey, + device_pubkey: Pubkey, + accesspass_pubkey: Pubkey, + /// Five activated multicast groups, split across feeds by the tests. + groups: Vec, + user_ip: Ipv4Addr, + user_pubkey: Pubkey, + user_tunnel_block: Pubkey, + multicast_publisher_block: Pubkey, + tunnel_ids: Pubkey, + dz_prefix_block: Pubkey, +} + +/// GlobalState/Config, Location, Exchange, Contributor, an Activated Device, five Activated +/// MulticastGroups, and an EdgeSeat access pass with no feeds yet. +async fn setup(client_ip: [u8; 4]) -> Fixture { + let (mut banks_client, payer, program_id, globalstate_pubkey, globalconfig_pubkey) = + setup_program_with_globalconfig().await; + let recent_blockhash = banks_client.get_latest_blockhash().await.unwrap(); + + let (user_tunnel_block, _, _) = + get_resource_extension_pda(&program_id, ResourceType::UserTunnelBlock); + let (multicast_publisher_block, _, _) = + get_resource_extension_pda(&program_id, ResourceType::MulticastPublisherBlock); + + let (location_pubkey, exchange_pubkey, contributor_pubkey) = setup_device_prerequisites( + &mut banks_client, + recent_blockhash, + program_id, + globalstate_pubkey, + globalconfig_pubkey, + &payer, + ) + .await; + + let gs = get_globalstate(&mut banks_client, globalstate_pubkey).await; + let (device_pubkey, _) = get_device_pda(&program_id, gs.account_index + 1); + let (tunnel_ids, _, _) = + get_resource_extension_pda(&program_id, ResourceType::TunnelIds(device_pubkey, 0)); + let (dz_prefix_block, _, _) = + get_resource_extension_pda(&program_id, ResourceType::DzPrefixBlock(device_pubkey, 0)); + execute_transaction( + &mut banks_client, + recent_blockhash, + program_id, + DoubleZeroInstruction::CreateDevice(DeviceCreateArgs { + code: "test-dev".to_string(), + device_type: DeviceType::Hybrid, + public_ip: [100, 0, 0, 1].into(), + dz_prefixes: "110.1.0.0/24".parse().unwrap(), + metrics_publisher_pk: Pubkey::default(), + mgmt_vrf: "mgmt".to_string(), + desired_status: None, + resource_count: 2, + }), + vec![ + AccountMeta::new(device_pubkey, false), + AccountMeta::new(contributor_pubkey, false), + AccountMeta::new(location_pubkey, false), + AccountMeta::new(exchange_pubkey, false), + AccountMeta::new(globalstate_pubkey, false), + AccountMeta::new(globalconfig_pubkey, false), + AccountMeta::new(tunnel_ids, false), + AccountMeta::new(dz_prefix_block, false), + ], + &payer, + ) + .await; + + execute_transaction( + &mut banks_client, + recent_blockhash, + program_id, + DoubleZeroInstruction::UpdateDevice(DeviceUpdateArgs { + max_users: Some(128), + ..DeviceUpdateArgs::default() + }), + vec![ + AccountMeta::new(device_pubkey, false), + AccountMeta::new(contributor_pubkey, false), + AccountMeta::new(location_pubkey, false), + AccountMeta::new(location_pubkey, false), + AccountMeta::new(globalstate_pubkey, false), + ], + &payer, + ) + .await; + + let mut groups = Vec::new(); + for i in 0..5 { + let gs = get_globalstate(&mut banks_client, globalstate_pubkey).await; + let (mgroup_pubkey, _) = get_multicastgroup_pda(&program_id, gs.account_index + 1); + execute_transaction( + &mut banks_client, + recent_blockhash, + program_id, + DoubleZeroInstruction::CreateMulticastGroup(MulticastGroupCreateArgs { + code: format!("group{i}"), + max_bandwidth: 1000, + owner: payer.pubkey(), + use_onchain_allocation: true, + }), + vec![ + AccountMeta::new(mgroup_pubkey, false), + AccountMeta::new(globalstate_pubkey, false), + AccountMeta::new( + get_resource_extension_pda(&program_id, ResourceType::MulticastGroupBlock).0, + false, + ), + ], + &payer, + ) + .await; + groups.push(mgroup_pubkey); + } + + // EdgeSeat passes are issued at the dynamic (0.0.0.0) PDA so one pass serves every machine the + // buyer connects, which is what lets two users share a feed's seats below. + let user_ip: Ipv4Addr = client_ip.into(); + let (accesspass_pubkey, _) = + get_accesspass_pda(&program_id, &Ipv4Addr::UNSPECIFIED, &payer.pubkey()); + execute_transaction( + &mut banks_client, + recent_blockhash, + program_id, + DoubleZeroInstruction::SetAccessPass(SetAccessPassArgs { + accesspass_type: AccessPassType::EdgeSeat(vec![]), + client_ip: Ipv4Addr::UNSPECIFIED, + last_access_epoch: 9999, + allow_multiple_ip: true, + max_unicast_users: 1, + max_multicast_users: 4, + }), + vec![ + AccountMeta::new(accesspass_pubkey, false), + AccountMeta::new(globalstate_pubkey, false), + AccountMeta::new(payer.pubkey(), false), + ], + &payer, + ) + .await; + + let (user_pubkey, _) = get_user_pda(&program_id, &user_ip, UserType::Multicast); + + Fixture { + banks_client, + payer, + program_id, + globalstate_pubkey, + exchange_pubkey, + device_pubkey, + accesspass_pubkey, + groups, + user_ip, + user_pubkey, + user_tunnel_block, + multicast_publisher_block, + tunnel_ids, + dz_prefix_block, + } +} + +async fn create_feed(f: &mut Fixture, code: &str, exchange: Pubkey, groups: Vec) -> Pubkey { + let (feed_pubkey, _) = get_feed_pda(&f.program_id, code, &exchange); + let recent_blockhash = f.banks_client.get_latest_blockhash().await.unwrap(); + execute_transaction( + &mut f.banks_client, + recent_blockhash, + f.program_id, + DoubleZeroInstruction::CreateFeed(FeedCreateArgs { + code: code.to_string(), + name: code.to_string(), + exchange, + groups, + }), + vec![ + AccountMeta::new(feed_pubkey, false), + AccountMeta::new(f.globalstate_pubkey, false), + ], + &f.payer, + ) + .await; + feed_pubkey +} + +/// Activated multicast groups beyond the five the fixture creates. +async fn create_groups(f: &mut Fixture, n: usize) -> Vec { + let recent_blockhash = wait_for_new_blockhash(&mut f.banks_client).await; + let mut groups = Vec::with_capacity(n); + for i in 0..n { + let gs = get_globalstate(&mut f.banks_client, f.globalstate_pubkey).await; + let (mgroup_pubkey, _) = get_multicastgroup_pda(&f.program_id, gs.account_index + 1); + execute_transaction( + &mut f.banks_client, + recent_blockhash, + f.program_id, + DoubleZeroInstruction::CreateMulticastGroup(MulticastGroupCreateArgs { + code: format!("extra{i}"), + max_bandwidth: 1000, + owner: f.payer.pubkey(), + use_onchain_allocation: true, + }), + vec![ + AccountMeta::new(mgroup_pubkey, false), + AccountMeta::new(f.globalstate_pubkey, false), + AccountMeta::new( + get_resource_extension_pda(&f.program_id, ResourceType::MulticastGroupBlock).0, + false, + ), + ], + &f.payer, + ) + .await; + groups.push(mgroup_pubkey); + } + groups +} + +fn seat(feed_key: Pubkey, max_users: u8) -> FeedSeat { + FeedSeat { + feed_key, + max_users, + max_future_users: max_users, + current_users: 0, + anniversary_day: 1, + window_end: TEST_WINDOW_END, + terminates_at: TEST_TERMINATES_AT, + } +} + +async fn set_pass_feeds(f: &mut Fixture, seats: Vec) { + let recent_blockhash = wait_for_new_blockhash(&mut f.banks_client).await; + let mut accounts = vec![ + AccountMeta::new(f.accesspass_pubkey, false), + AccountMeta::new(f.globalstate_pubkey, false), + ]; + for s in &seats { + accounts.push(AccountMeta::new(s.feed_key, false)); + } + let mut tx = create_transaction( + f.program_id, + &DoubleZeroInstruction::SetAccessPassFeeds(SetAccessPassFeedsArgs { + client_ip: Ipv4Addr::UNSPECIFIED, + user_payer: f.payer.pubkey(), + feeds: seats + .iter() + .map(|s| FeedSeatConfig { + max_users: s.max_users, + max_future_users: s.max_future_users, + anniversary_day: s.anniversary_day, + window_end: s.window_end, + terminates_at: s.terminates_at, + }) + .collect(), + }), + &accounts, + &f.payer, + ); + tx.try_sign(&[&f.payer], recent_blockhash).unwrap(); + f.banks_client.process_transaction(tx).await.unwrap(); +} + +/// Bring the Multicast user into existence on `feed` joined to `group`. `UpdateFeedSubscription` +/// operates on an existing user, and until `CreateUser` is made naked (#4110) this is the only way +/// to create one under an EdgeSeat pass. +async fn create_user_on(f: &mut Fixture, feed: Pubkey, group: Pubkey) { + let ip = f.user_ip; + create_user_at(f, ip, feed, group).await +} + +/// Same, for an arbitrary client IP — a second machine under the same dynamic pass. +async fn create_user_at(f: &mut Fixture, ip: Ipv4Addr, feed: Pubkey, group: Pubkey) { + let (user_pubkey, _) = get_user_pda(&f.program_id, &ip, UserType::Multicast); + let recent_blockhash = wait_for_new_blockhash(&mut f.banks_client).await; + let accounts = vec![ + AccountMeta::new(user_pubkey, false), + AccountMeta::new(f.device_pubkey, false), + AccountMeta::new(group, false), + AccountMeta::new(f.accesspass_pubkey, false), + AccountMeta::new(f.globalstate_pubkey, false), + AccountMeta::new(f.user_tunnel_block, false), + AccountMeta::new(f.multicast_publisher_block, false), + AccountMeta::new(f.tunnel_ids, false), + AccountMeta::new(f.dz_prefix_block, false), + AccountMeta::new_readonly(feed, false), + ]; + let mut tx = create_transaction_with_extra_accounts( + f.program_id, + &DoubleZeroInstruction::CreateSubscribeUser(UserCreateSubscribeArgs { + user_type: UserType::Multicast, + cyoa_type: UserCYOA::GREOverDIA, + client_ip: ip, + publisher: false, + subscriber: true, + tunnel_endpoint: Ipv4Addr::UNSPECIFIED, + dz_prefix_count: 1, + owner: Pubkey::default(), + }), + &accounts, + &f.payer, + &[], + ); + tx.try_sign(&[&f.payer], recent_blockhash).unwrap(); + f.banks_client.process_transaction(tx).await.unwrap(); +} + +#[allow(clippy::too_many_arguments)] +async fn try_join_as( + f: &mut Fixture, + user_pubkey: Pubkey, + device: Pubkey, + feeds: &[Pubkey], + groups: &[Pubkey], +) -> Result<(), BanksClientError> { + let recent_blockhash = wait_for_new_blockhash(&mut f.banks_client).await; + let mut accounts = vec![ + AccountMeta::new(f.accesspass_pubkey, false), + AccountMeta::new(user_pubkey, false), + AccountMeta::new(f.globalstate_pubkey, false), + AccountMeta::new_readonly(device, false), + ]; + accounts.extend(feeds.iter().map(|k| AccountMeta::new_readonly(*k, false))); + accounts.extend(groups.iter().map(|k| AccountMeta::new(*k, false))); + + let mut tx = create_transaction_with_extra_accounts( + f.program_id, + &DoubleZeroInstruction::SubscribeFeed(SubscribeFeedArgs { + feed_count: feeds.len() as u8, + }), + &accounts, + &f.payer, + &[], + ); + tx.try_sign(&[&f.payer], recent_blockhash).unwrap(); + f.banks_client.process_transaction(tx).await +} + +async fn join( + f: &mut Fixture, + feeds: &[Pubkey], + groups: &[Pubkey], +) -> Result<(), BanksClientError> { + let (user, device) = (f.user_pubkey, f.device_pubkey); + try_join_as(f, user, device, feeds, groups).await +} + +#[allow(clippy::too_many_arguments)] +async fn try_leave_as( + f: &mut Fixture, + user_pubkey: Pubkey, + device: Pubkey, + targets: &[Pubkey], + retained: &[Pubkey], + groups: &[Pubkey], +) -> Result<(), BanksClientError> { + let recent_blockhash = wait_for_new_blockhash(&mut f.banks_client).await; + let mut accounts = vec![ + AccountMeta::new(f.accesspass_pubkey, false), + AccountMeta::new(user_pubkey, false), + AccountMeta::new(f.globalstate_pubkey, false), + AccountMeta::new_readonly(device, false), + ]; + accounts.extend( + targets + .iter() + .chain(retained) + .map(|k| AccountMeta::new_readonly(*k, false)), + ); + accounts.extend(groups.iter().map(|k| AccountMeta::new(*k, false))); + + let mut tx = create_transaction_with_extra_accounts( + f.program_id, + &DoubleZeroInstruction::UnsubscribeFeed(UnsubscribeFeedArgs { + feed_count: targets.len() as u8, + retained_feed_count: retained.len() as u8, + }), + &accounts, + &f.payer, + &[], + ); + tx.try_sign(&[&f.payer], recent_blockhash).unwrap(); + f.banks_client.process_transaction(tx).await +} + +async fn leave( + f: &mut Fixture, + targets: &[Pubkey], + retained: &[Pubkey], + groups: &[Pubkey], +) -> Result<(), BanksClientError> { + let (user, device) = (f.user_pubkey, f.device_pubkey); + try_leave_as(f, user, device, targets, retained, groups).await +} + +/// An IBRL user under the same pass, for the user-type gate. +async fn create_ibrl_user(f: &mut Fixture, ip: Ipv4Addr) { + let (user_pubkey, _) = get_user_pda(&f.program_id, &ip, UserType::IBRL); + let recent_blockhash = wait_for_new_blockhash(&mut f.banks_client).await; + execute_transaction( + &mut f.banks_client, + recent_blockhash, + f.program_id, + DoubleZeroInstruction::CreateUser(UserCreateArgs { + client_ip: ip, + user_type: UserType::IBRL, + cyoa_type: UserCYOA::GREOverDIA, + tunnel_endpoint: Ipv4Addr::UNSPECIFIED, + dz_prefix_count: 1, + }), + vec![ + AccountMeta::new(user_pubkey, false), + AccountMeta::new(f.device_pubkey, false), + AccountMeta::new(f.accesspass_pubkey, false), + AccountMeta::new(f.globalstate_pubkey, false), + AccountMeta::new(f.user_tunnel_block, false), + AccountMeta::new(f.multicast_publisher_block, false), + AccountMeta::new(f.tunnel_ids, false), + AccountMeta::new(f.dz_prefix_block, false), + ], + &f.payer, + ) + .await; +} + +async fn read_pass(f: &mut Fixture) -> AccessPass { + get_account_data(&mut f.banks_client, f.accesspass_pubkey) + .await + .expect("access pass") + .get_accesspass() + .unwrap() +} + +async fn read_user(f: &mut Fixture) -> User { + get_account_data(&mut f.banks_client, f.user_pubkey) + .await + .expect("user") + .get_user() + .unwrap() +} + +/// Match the error structurally rather than on its debug text, so a test cannot pass because some +/// other instruction in the transaction failed or because the formatting changed. +fn assert_custom_error(err: &BanksClientError, code: u32) { + match err { + BanksClientError::TransactionError(TransactionError::InstructionError( + 0, + InstructionError::Custom(actual), + )) if *actual == code => {} + other => panic!("expected Custom({code}), got {other:?}"), + } +} + +fn seat_users(pass: &AccessPass, feed: &Pubkey) -> u8 { + pass.feed_seats() + .iter() + .find(|s| &s.feed_key == feed) + .expect("seat for feed") + .current_users +} + +// A feed's whole group set is joined in one transaction, and the three groups inside it cost a +// single seat. The seat is capped at 1, so a per-group tick would fail as FeedSeatFull. +#[tokio::test] +async fn test_feed_subscription_joins_every_group_for_one_seat() { + let mut f = setup([100, 0, 0, 20]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let feed = create_feed(&mut f, "feed1", exchange, vec![g[0], g[1], g[2]]).await; + set_pass_feeds(&mut f, vec![seat(feed, 1)]).await; + + create_user_on(&mut f, feed, g[0]).await; + join(&mut f, &[feed], &[g[1], g[2]]).await.unwrap(); + + let user = read_user(&mut f).await; + assert_eq!(user.subscribers, vec![g[0], g[1], g[2]]); + assert_eq!(user.feed_pks, vec![feed]); + assert_eq!(seat_users(&read_pass(&mut f).await, &feed), 1); +} + +// A group drawn from a second feed takes that feed's own seat, and leaves the first untouched. +#[tokio::test] +async fn test_second_feed_takes_its_own_seat() { + let mut f = setup([100, 0, 0, 21]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let feed1 = create_feed(&mut f, "feed1", exchange, vec![g[0], g[1]]).await; + let feed2 = create_feed(&mut f, "feed2", exchange, vec![g[2], g[3]]).await; + set_pass_feeds(&mut f, vec![seat(feed1, 1), seat(feed2, 1)]).await; + + create_user_on(&mut f, feed1, g[0]).await; + join(&mut f, &[feed2], &[g[2], g[3]]).await.unwrap(); + + let user = read_user(&mut f).await; + assert_eq!(user.feed_pks, vec![feed1, feed2]); + let pass = read_pass(&mut f).await; + assert_eq!(seat_users(&pass, &feed1), 1); + assert_eq!(seat_users(&pass, &feed2), 1); +} + +// Both feeds' groups join in a single transaction, taking one seat each. +#[tokio::test] +async fn test_two_feeds_in_one_transaction() { + let mut f = setup([100, 0, 0, 22]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let feed1 = create_feed(&mut f, "feed1", exchange, vec![g[0], g[1]]).await; + let feed2 = create_feed(&mut f, "feed2", exchange, vec![g[2], g[3]]).await; + set_pass_feeds(&mut f, vec![seat(feed1, 1), seat(feed2, 1)]).await; + + create_user_on(&mut f, feed1, g[0]).await; + join(&mut f, &[feed1, feed2], &[g[1], g[2], g[3]]) + .await + .unwrap(); + + let user = read_user(&mut f).await; + assert_eq!(user.subscribers, vec![g[0], g[1], g[2], g[3]]); + let pass = read_pass(&mut f).await; + assert_eq!(seat_users(&pass, &feed1), 1); + assert_eq!(seat_users(&pass, &feed2), 1); +} + +// Dropping the user's last group in a feed releases that feed's seat. +#[tokio::test] +async fn test_removal_releases_the_seat_on_the_last_group() { + let mut f = setup([100, 0, 0, 23]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let feed = create_feed(&mut f, "feed1", exchange, vec![g[0], g[1]]).await; + set_pass_feeds(&mut f, vec![seat(feed, 1)]).await; + + create_user_on(&mut f, feed, g[0]).await; + join(&mut f, &[feed], &[g[1]]).await.unwrap(); + leave(&mut f, &[feed], &[], &[g[0], g[1]]).await.unwrap(); + + let user = read_user(&mut f).await; + assert!(user.subscribers.is_empty()); + assert!(user.feed_pks.is_empty()); + assert_eq!(seat_users(&read_pass(&mut f).await, &feed), 0); +} + +// Leaving one feed must not drop a group a retained feed still carries. Alice keeps feed1, which +// also sells g0, so leaving feed2 costs her g1 only and leaves feed1's seat charged. +#[tokio::test] +async fn test_leaving_a_feed_keeps_a_group_a_retained_feed_covers() { + let mut f = setup([100, 0, 0, 24]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let feed1 = create_feed(&mut f, "feed1", exchange, vec![g[0]]).await; + let feed2 = create_feed(&mut f, "feed2", exchange, vec![g[0], g[1]]).await; + set_pass_feeds(&mut f, vec![seat(feed1, 1), seat(feed2, 1)]).await; + + create_user_on(&mut f, feed1, g[0]).await; + join(&mut f, &[feed2], &[g[1]]).await.unwrap(); + + // g[0] is covered by feed1, so only g[1] departs with feed2. + leave(&mut f, &[feed2], &[feed1], &[g[1]]).await.unwrap(); + + let user = read_user(&mut f).await; + assert_eq!(user.subscribers, vec![g[0]]); + assert_eq!(user.feed_pks, vec![feed1]); + let pass = read_pass(&mut f).await; + assert_eq!(seat_users(&pass, &feed1), 1); + assert_eq!(seat_users(&pass, &feed2), 0); +} + +// A leave that omits a feed the user still holds is rejected rather than stranding its seat. +#[tokio::test] +async fn test_leave_omitting_a_held_feed_rejected() { + let mut f = setup([100, 0, 0, 32]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let feed1 = create_feed(&mut f, "feed1", exchange, vec![g[0]]).await; + let feed2 = create_feed(&mut f, "feed2", exchange, vec![g[0], g[1]]).await; + set_pass_feeds(&mut f, vec![seat(feed1, 1), seat(feed2, 1)]).await; + + create_user_on(&mut f, feed1, g[0]).await; + join(&mut f, &[feed2], &[g[1]]).await.unwrap(); + + let err = leave(&mut f, &[feed2], &[], &[g[1]]).await.unwrap_err(); + assert_custom_error(&err, 92); +} + +// The group list must be exactly what the target feeds change, so a stale client cannot half-apply. +#[tokio::test] +async fn test_group_list_not_matching_the_feeds_rejected() { + let mut f = setup([100, 0, 0, 25]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let feed = create_feed(&mut f, "feed1", exchange, vec![g[0], g[1]]).await; + set_pass_feeds(&mut f, vec![seat(feed, 1)]).await; + create_user_on(&mut f, feed, g[0]).await; + + // Same count as the real change set (g[1]), but the wrong group. + let err = join(&mut f, &[feed], &[g[4]]).await.unwrap_err(); + assert_custom_error(&err, 65); +} + +// A feed that is not provisioned on the pass is rejected: FeedNotOnAccessPass (93). +#[tokio::test] +async fn test_feed_not_on_the_pass_rejected() { + let mut f = setup([100, 0, 0, 26]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let feed = create_feed(&mut f, "feed1", exchange, vec![g[0], g[1]]).await; + let unprovisioned = create_feed(&mut f, "feed2", exchange, vec![g[2]]).await; + set_pass_feeds(&mut f, vec![seat(feed, 1)]).await; + create_user_on(&mut f, feed, g[0]).await; + + let err = join(&mut f, &[unprovisioned], &[g[2]]).await.unwrap_err(); + assert_custom_error(&err, 93); +} + +// A feed serving a different metro than the user's device is rejected: MetroMismatch (91). +#[tokio::test] +async fn test_feed_serving_another_metro_rejected() { + let mut f = setup([100, 0, 0, 27]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let feed = create_feed(&mut f, "feed1", exchange, vec![g[0], g[1]]).await; + // A feed keyed to an exchange the device does not sit in. + let other_metro = create_feed(&mut f, "feed2", Pubkey::new_unique(), vec![g[2]]).await; + set_pass_feeds(&mut f, vec![seat(feed, 1), seat(other_metro, 1)]).await; + create_user_on(&mut f, feed, g[0]).await; + + let err = join(&mut f, &[other_metro], &[g[2]]).await.unwrap_err(); + assert_custom_error(&err, 91); +} + +// A device that is not the user's is rejected: UserDeviceMismatch (102). Without this a caller +// could pass any device and have its exchange satisfy the metro check. +#[tokio::test] +async fn test_foreign_device_rejected() { + let mut f = setup([100, 0, 0, 28]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let feed = create_feed(&mut f, "feed1", exchange, vec![g[0], g[1]]).await; + set_pass_feeds(&mut f, vec![seat(feed, 1)]).await; + create_user_on(&mut f, feed, g[0]).await; + + let (foreign, user) = (f.globalstate_pubkey, f.user_pubkey); + let err = try_join_as(&mut f, user, foreign, &[feed], &[g[1]]) + .await + .unwrap_err(); + assert_custom_error(&err, 102); +} + +// Two machines share the pass. A feed sold for one user admits the first and rejects the second +// with FeedSeatFull (95) — the cap is per feed, and this is the boundary a buyer actually hits. +#[tokio::test] +async fn test_seat_cap_rejects_a_second_machine() { + let mut f = setup([100, 0, 0, 29]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + // `entry` has room for both machines and only exists so each user account can be created; until + // CreateUser goes naked (#4110) creating a user under an EdgeSeat pass always takes a seat. + // `scarce` is the feed under test, sold for a single user. + let entry = create_feed(&mut f, "entry", exchange, vec![g[0]]).await; + let scarce = create_feed(&mut f, "scarce", exchange, vec![g[2]]).await; + set_pass_feeds(&mut f, vec![seat(entry, 2), seat(scarce, 1)]).await; + + let second_ip: Ipv4Addr = [100, 0, 0, 60].into(); + let (second_user, _) = get_user_pda(&f.program_id, &second_ip, UserType::Multicast); + create_user_on(&mut f, entry, g[0]).await; + create_user_at(&mut f, second_ip, entry, g[0]).await; + assert_eq!(seat_users(&read_pass(&mut f).await, &entry), 2); + + // Machine 1 takes the scarce feed's only seat. + join(&mut f, &[scarce], &[g[2]]).await.unwrap(); + assert_eq!(seat_users(&read_pass(&mut f).await, &scarce), 1); + + // Machine 2 is a legitimate user on the same pass, but the feed is sold out. + let device = f.device_pubkey; + let err = try_join_as(&mut f, second_user, device, &[scarce], &[g[2]]) + .await + .unwrap_err(); + assert_custom_error(&err, 95); + assert_eq!(seat_users(&read_pass(&mut f).await, &scarce), 1); +} + +// The comped path still works on an EdgeSeat pass: a group a foundation member put on the +// subscriber allowlist is joinable through UpdateMulticastGroupRoles, and takes no feed seat. +#[tokio::test] +async fn test_allowlisted_group_joins_without_a_seat() { + let mut f = setup([100, 0, 0, 30]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let feed = create_feed(&mut f, "feed1", exchange, vec![g[0]]).await; + set_pass_feeds(&mut f, vec![seat(feed, 1)]).await; + create_user_on(&mut f, feed, g[0]).await; + + // Foundation comps g[4], which no feed carries. + let recent_blockhash = wait_for_new_blockhash(&mut f.banks_client).await; + execute_transaction( + &mut f.banks_client, + recent_blockhash, + f.program_id, + DoubleZeroInstruction::AddMulticastGroupSubAllowlist( + doublezero_serviceability::processors::multicastgroup::allowlist::subscriber::add::AddMulticastGroupSubAllowlistArgs { + client_ip: f.user_ip, + user_payer: f.payer.pubkey(), + }, + ), + vec![ + AccountMeta::new(g[4], false), + AccountMeta::new(f.accesspass_pubkey, false), + AccountMeta::new(f.globalstate_pubkey, false), + AccountMeta::new(f.payer.pubkey(), false), + ], + &f.payer, + ) + .await; + + let recent_blockhash = wait_for_new_blockhash(&mut f.banks_client).await; + let mut tx = create_transaction_with_extra_accounts( + f.program_id, + &DoubleZeroInstruction::UpdateMulticastGroupRoles(UpdateMulticastGroupRolesArgs { + client_ip: f.user_ip, + publisher: false, + subscriber: true, + use_onchain_allocation: true, + }), + &vec![ + AccountMeta::new(g[4], false), + AccountMeta::new(f.accesspass_pubkey, false), + AccountMeta::new(f.user_pubkey, false), + AccountMeta::new(f.globalstate_pubkey, false), + AccountMeta::new(f.multicast_publisher_block, false), + ], + &f.payer, + &[], + ); + tx.try_sign(&[&f.payer], recent_blockhash).unwrap(); + f.banks_client.process_transaction(tx).await.unwrap(); + + let user = read_user(&mut f).await; + assert_eq!(user.subscribers, vec![g[0], g[4]]); + // The comped group belongs to no feed, so it consumed nothing. + assert_eq!(user.feed_pks, vec![feed]); + assert_eq!(seat_users(&read_pass(&mut f).await, &feed), 1); +} + +// The hole this PR closes: an EdgeSeat holder may no longer reach a group through +// UpdateMulticastGroupRoles just because a feed carries it. That path is now allowlist-only, so a +// purchased group must go through UpdateFeedSubscription and charge its seat. NotAllowed (8). +#[tokio::test] +async fn test_feed_group_not_joinable_through_the_roles_instruction() { + let mut f = setup([100, 0, 0, 31]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let feed = create_feed(&mut f, "feed1", exchange, vec![g[0], g[1]]).await; + set_pass_feeds(&mut f, vec![seat(feed, 1)]).await; + create_user_on(&mut f, feed, g[0]).await; + + let recent_blockhash = wait_for_new_blockhash(&mut f.banks_client).await; + let mut tx = create_transaction_with_extra_accounts( + f.program_id, + &DoubleZeroInstruction::UpdateMulticastGroupRoles(UpdateMulticastGroupRolesArgs { + client_ip: f.user_ip, + publisher: false, + subscriber: true, + use_onchain_allocation: true, + }), + &vec![ + AccountMeta::new(g[1], false), + AccountMeta::new(f.accesspass_pubkey, false), + AccountMeta::new(f.user_pubkey, false), + AccountMeta::new(f.globalstate_pubkey, false), + AccountMeta::new(f.multicast_publisher_block, false), + ], + &f.payer, + &[], + ); + tx.try_sign(&[&f.payer], recent_blockhash).unwrap(); + let err = f.banks_client.process_transaction(tx).await.unwrap_err(); + assert_custom_error(&err, 8); +} + +// This instruction is EdgeSeat-only: a Prepaid pass carries no feeds and must use the allowlist path. +#[tokio::test] +async fn test_non_edgeseat_pass_rejected() { + let mut f = setup([100, 0, 0, 33]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let feed = create_feed(&mut f, "feed1", exchange, vec![g[0], g[1]]).await; + set_pass_feeds(&mut f, vec![seat(feed, 1)]).await; + create_user_on(&mut f, feed, g[0]).await; + + // Downgrade the pass to Prepaid, keeping the same PDA and user. + let recent_blockhash = wait_for_new_blockhash(&mut f.banks_client).await; + execute_transaction( + &mut f.banks_client, + recent_blockhash, + f.program_id, + DoubleZeroInstruction::SetAccessPass(SetAccessPassArgs { + accesspass_type: AccessPassType::Prepaid, + client_ip: Ipv4Addr::UNSPECIFIED, + last_access_epoch: 9999, + allow_multiple_ip: true, + max_unicast_users: 1, + max_multicast_users: 4, + }), + vec![ + AccountMeta::new(f.accesspass_pubkey, false), + AccountMeta::new(f.globalstate_pubkey, false), + AccountMeta::new(f.payer.pubkey(), false), + ], + &f.payer, + ) + .await; + + let err = join(&mut f, &[feed], &[g[1]]).await.unwrap_err(); + assert_custom_error(&err, 101); +} + +// Only a Multicast user occupies a feed seat, so no other user type may hold a feed subscription. +#[tokio::test] +async fn test_non_multicast_user_rejected() { + let mut f = setup([100, 0, 0, 34]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let feed = create_feed(&mut f, "feed1", exchange, vec![g[0], g[1]]).await; + set_pass_feeds(&mut f, vec![seat(feed, 2)]).await; + create_user_on(&mut f, feed, g[0]).await; + + // An IBRL user at another IP under the same dynamic pass. + let ibrl_ip: Ipv4Addr = [100, 0, 0, 70].into(); + let (ibrl_user, _) = get_user_pda(&f.program_id, &ibrl_ip, UserType::IBRL); + create_ibrl_user(&mut f, ibrl_ip).await; + + let device = f.device_pubkey; + let err = try_join_as(&mut f, ibrl_user, device, &[feed], &[g[1]]) + .await + .unwrap_err(); + assert_custom_error(&err, 104); +} + +// feed_count == 0 leaves nothing to derive the group set from. +#[tokio::test] +async fn test_zero_feed_count_rejected() { + let mut f = setup([100, 0, 0, 35]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let feed = create_feed(&mut f, "feed1", exchange, vec![g[0], g[1]]).await; + set_pass_feeds(&mut f, vec![seat(feed, 1)]).await; + create_user_on(&mut f, feed, g[0]).await; + + let err = join(&mut f, &[], &[g[1]]).await.unwrap_err(); + assert_custom_error(&err, 65); +} + +// The same feed twice would double-count its seat. +#[tokio::test] +async fn test_duplicate_feed_rejected() { + let mut f = setup([100, 0, 0, 36]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let feed = create_feed(&mut f, "feed1", exchange, vec![g[0], g[1]]).await; + set_pass_feeds(&mut f, vec![seat(feed, 1)]).await; + create_user_on(&mut f, feed, g[0]).await; + + let err = join(&mut f, &[feed, feed], &[g[1]]).await.unwrap_err(); + assert_custom_error(&err, 65); +} + +// Naming the departing feed as retained would make every one of its groups look covered, so nothing +// is unsubscribed while the seat is still released. +#[tokio::test] +async fn test_target_passed_as_retained_rejected() { + let mut f = setup([100, 0, 0, 38]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let feed = create_feed(&mut f, "feed1", exchange, vec![g[0], g[1]]).await; + set_pass_feeds(&mut f, vec![seat(feed, 1)]).await; + create_user_on(&mut f, feed, g[0]).await; + join(&mut f, &[feed], &[g[1]]).await.unwrap(); + + let err = leave(&mut f, &[feed], &[feed], &[]).await.unwrap_err(); + assert_custom_error(&err, 65); + + // The seat and both subscriptions survive the rejected call. + let user = read_user(&mut f).await; + assert_eq!(user.subscribers, vec![g[0], g[1]]); + assert_eq!(seat_users(&read_pass(&mut f).await, &feed), 1); +} + +// A feed the user never held cannot stand in as retained: overlapping groups would survive the leave +// while the target's seat is released. +#[tokio::test] +async fn test_unheld_feed_as_retained_rejected() { + let mut f = setup([100, 0, 0, 39]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let held = create_feed(&mut f, "held", exchange, vec![g[0], g[1]]).await; + let unheld = create_feed(&mut f, "unheld", exchange, vec![g[0], g[1]]).await; + set_pass_feeds(&mut f, vec![seat(held, 1), seat(unheld, 1)]).await; + create_user_on(&mut f, held, g[0]).await; + join(&mut f, &[held], &[g[1]]).await.unwrap(); + + let err = leave(&mut f, &[held], &[unheld], &[]).await.unwrap_err(); + assert_custom_error(&err, 65); + + let user = read_user(&mut f).await; + assert_eq!(user.subscribers, vec![g[0], g[1]]); + assert_eq!(seat_users(&read_pass(&mut f).await, &held), 1); +} + +// The pass admin drops a feed the user still holds (SetAccessPassFeeds never refuses that). The +// stale feed has no seat left to strand, so it no longer has to be named: the user can still leave +// their other feeds, and the stale entry is pruned along the way. +#[tokio::test] +async fn test_leave_succeeds_after_the_pass_drops_a_held_feed() { + let mut f = setup([100, 0, 0, 41]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let feed1 = create_feed(&mut f, "feed1", exchange, vec![g[0]]).await; + let feed2 = create_feed(&mut f, "feed2", exchange, vec![g[1]]).await; + set_pass_feeds(&mut f, vec![seat(feed1, 1), seat(feed2, 1)]).await; + create_user_on(&mut f, feed1, g[0]).await; + join(&mut f, &[feed2], &[g[1]]).await.unwrap(); + + set_pass_feeds(&mut f, vec![seat(feed1, 1)]).await; + + leave(&mut f, &[feed1], &[], &[g[0]]).await.unwrap(); + + let user = read_user(&mut f).await; + assert!(user.feed_pks.is_empty()); + // The stale feed's group stays subscribed until reconciled (or feed2 is named as a target). + assert_eq!(user.subscribers, vec![g[1]]); + assert_eq!(seat_users(&read_pass(&mut f).await, &feed1), 0); +} + +// A stale feed named as a target unsubscribes its groups, still honoring retained coverage, and +// touches no seat. +#[tokio::test] +async fn test_stale_feed_named_as_target_cleans_up() { + let mut f = setup([100, 0, 0, 42]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let feed1 = create_feed(&mut f, "feed1", exchange, vec![g[0]]).await; + let feed2 = create_feed(&mut f, "feed2", exchange, vec![g[0], g[1]]).await; + set_pass_feeds(&mut f, vec![seat(feed1, 1), seat(feed2, 1)]).await; + create_user_on(&mut f, feed1, g[0]).await; + join(&mut f, &[feed2], &[g[1]]).await.unwrap(); + + set_pass_feeds(&mut f, vec![seat(feed1, 1)]).await; + + // g[0] stays covered by retained feed1; only g[1] departs with the stale feed2. + leave(&mut f, &[feed2], &[feed1], &[g[1]]).await.unwrap(); + + let user = read_user(&mut f).await; + assert_eq!(user.subscribers, vec![g[0]]); + assert_eq!(user.feed_pks, vec![feed1]); + assert_eq!(seat_users(&read_pass(&mut f).await, &feed1), 1); +} + +// A stale feed carries no entitlement, so it cannot stand in as retained coverage. +#[tokio::test] +async fn test_stale_feed_as_retained_rejected() { + let mut f = setup([100, 0, 0, 43]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + let feed1 = create_feed(&mut f, "feed1", exchange, vec![g[0]]).await; + let feed2 = create_feed(&mut f, "feed2", exchange, vec![g[0], g[1]]).await; + set_pass_feeds(&mut f, vec![seat(feed1, 1), seat(feed2, 1)]).await; + create_user_on(&mut f, feed1, g[0]).await; + join(&mut f, &[feed2], &[g[1]]).await.unwrap(); + + set_pass_feeds(&mut f, vec![seat(feed2, 1)]).await; + + let err = leave(&mut f, &[feed2], &[feed1], &[g[1]]) + .await + .unwrap_err(); + assert_custom_error(&err, 93); +} + +// MAX_FEED_GROUPS is bounded by transaction capacity: a feed at the cap joins in one transaction. +#[tokio::test] +async fn test_feed_at_max_groups_joins_in_one_transaction() { + let mut f = setup([100, 0, 0, 44]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + // `entry` only exists so the user account can be created (see create_user_on). + let entry = create_feed(&mut f, "entry", exchange, vec![g[0]]).await; + let mut big_groups = g[1..5].to_vec(); + let extra = create_groups(&mut f, MAX_FEED_GROUPS - big_groups.len()).await; + big_groups.extend(extra); + let big = create_feed(&mut f, "big", exchange, big_groups.clone()).await; + set_pass_feeds(&mut f, vec![seat(entry, 1), seat(big, 1)]).await; + create_user_on(&mut f, entry, g[0]).await; + + join(&mut f, &[big], &big_groups).await.unwrap(); + + let user = read_user(&mut f).await; + assert_eq!(user.subscribers.len(), 1 + MAX_FEED_GROUPS); + assert_eq!(seat_users(&read_pass(&mut f).await, &big), 1); +} + +// A join that would take the user past MAX_USER_FEEDS is refused: UserFeedLimitExceeded (103). +#[tokio::test] +async fn test_join_past_the_user_feed_cap_rejected() { + let mut f = setup([100, 0, 0, 45]).await; + let (exchange, g) = (f.exchange_pubkey, f.groups.clone()); + // Every feed carries the same group, so each join costs its seat and adds no groups. + let mut feeds = Vec::new(); + for i in 0..=MAX_USER_FEEDS { + feeds.push(create_feed(&mut f, &format!("feed{i}"), exchange, vec![g[0]]).await); + } + let seats = feeds.iter().map(|k| seat(*k, 1)).collect(); + set_pass_feeds(&mut f, seats).await; + create_user_on(&mut f, feeds[0], g[0]).await; + + for feed in &feeds[1..MAX_USER_FEEDS] { + join(&mut f, &[*feed], &[]).await.unwrap(); + } + assert_eq!(read_user(&mut f).await.feed_pks.len(), MAX_USER_FEEDS); + + let err = join(&mut f, &[feeds[MAX_USER_FEEDS]], &[]) + .await + .unwrap_err(); + assert_custom_error(&err, 103); +} diff --git a/smartcontract/programs/doublezero-serviceability/tests/test_helpers.rs b/smartcontract/programs/doublezero-serviceability/tests/test_helpers.rs index 273a694dc9..0a427bbba8 100644 --- a/smartcontract/programs/doublezero-serviceability/tests/test_helpers.rs +++ b/smartcontract/programs/doublezero-serviceability/tests/test_helpers.rs @@ -635,8 +635,10 @@ pub async fn setup_program_with_globalconfig() -> (BanksClient, Keypair, Pubkey, processor!(process_instruction), ); // SetGlobalConfig creates multiple ResourceExtension accounts; raise the budget so this - // doesn't flake under load when many test processes run concurrently. - program_test.set_compute_max_units(1_000_000); + // doesn't flake under load when many test processes run concurrently. Set to the protocol max + // (mirrors MAX_COMPUTE_UNIT_LIMIT) so instructions that write many accounts in one transaction + // — e.g. UpdateFeedSubscription across several feeds and groups — also fit. + program_test.set_compute_max_units(1_400_000); let (mut banks_client, payer, recent_blockhash) = program_test.start().await; let (program_config_pubkey, _) = get_program_config_pda(&program_id);