diff --git a/actors/market/src/lib.rs b/actors/market/src/lib.rs index 5ebffff820..c6ff09ad6c 100644 --- a/actors/market/src/lib.rs +++ b/actors/market/src/lib.rs @@ -799,6 +799,15 @@ impl Actor { let proposals = st.load_proposals(rt.store())?; let states = st.load_deal_states(rt.store())?; + // Load the balance tables once too, rather than letting each slashed deal below + // independently load-and-reflush them: a sector can carry many deals, and most of + // that per-deal round trip is redundant when they all touch the same couple of + // tables. + let mut escrow_table = + BalanceTable::from_root(rt.store(), &st.escrow_table, "escrow table")?; + let mut locked_table = + BalanceTable::from_root(rt.store(), &st.locked_table, "locked table")?; + // The sector deals mapping is removed all at once. // Note there may be some deal states that are not removed here, // despite deletion of this mapping, e.g. for expired but not-yet-settled deals. @@ -862,7 +871,8 @@ impl Actor { } state.slash_epoch = params.epoch; - total_slashed += st.process_slashed_deal(rt.store(), &deal, &state)?; + total_slashed += + st.process_slashed_deal(&mut escrow_table, &mut locked_table, &deal, &state)?; st.remove_completed_deal(rt.store(), id)?; emit::deal_terminated( @@ -873,6 +883,9 @@ impl Actor { )?; } + st.escrow_table = escrow_table.root()?; + st.locked_table = locked_table.root()?; + Ok(total_slashed) })?; diff --git a/actors/market/src/state.rs b/actors/market/src/state.rs index 9cd01cc59b..eb5d8bd12b 100644 --- a/actors/market/src/state.rs +++ b/actors/market/src/state.rs @@ -919,9 +919,16 @@ impl State { Ok((TokenAmount::zero(), elapsed_payment, false, false)) } + /// Takes already-loaded balance tables rather than loading (and re-flushing) them from + /// `self.escrow_table`/`self.locked_table` internally. Callers processing many deals in one + /// transaction (e.g. sector termination) should load the tables once, call this once per + /// deal, and flush the tables once at the end -- each independent load/flush of a balance + /// table is a real HAMT round trip, and there's no need to pay for one per deal when most + /// of them touch the same one or two addresses. pub fn process_slashed_deal( &mut self, - store: &BS, + escrow_table: &mut BalanceTable, + locked_table: &mut BalanceTable, proposal: &DealProposal, state: &DealState, ) -> Result @@ -934,19 +941,30 @@ impl State { let num_epochs_elapsed = max(0, payment_end_epoch - payment_start_epoch); let total_payment = &proposal.storage_price_per_epoch * num_epochs_elapsed; if total_payment.is_positive() { - self.transfer_balance(store, &proposal.client, &proposal.provider, &total_payment)?; + self.transfer_balance_in_tables( + escrow_table, + locked_table, + &proposal.client, + &proposal.provider, + &total_payment, + )?; } // unlock client collateral and locked storage fee let payment_remaining = deal_get_payment_remaining(proposal, state.slash_epoch)?; // Unlock remaining storage fee - self.unlock_balance(store, &proposal.client, &payment_remaining, Reason::ClientStorageFee) - .context("unlocking client storage fee")?; + self.unlock_balance_in_table( + locked_table, + &proposal.client, + &payment_remaining, + Reason::ClientStorageFee, + ) + .context("unlocking client storage fee")?; // Unlock client collateral - self.unlock_balance( - store, + self.unlock_balance_in_table( + locked_table, &proposal.client, &proposal.client_collateral, Reason::ClientCollateral, @@ -955,8 +973,14 @@ impl State { // slash provider collateral let slashed = proposal.provider_collateral.clone(); - self.slash_balance(store, &proposal.provider, &slashed, Reason::ProviderCollateral) - .context("slashing balance")?; + self.slash_balance_in_tables( + escrow_table, + locked_table, + &proposal.provider, + &slashed, + Reason::ProviderCollateral, + ) + .context("slashing balance")?; Ok(slashed) } @@ -1097,9 +1121,11 @@ impl State { Ok(()) } - fn unlock_balance( + /// Takes an already-loaded locked-balance table instead of loading (and re-flushing) it + /// from `self.locked_table` internally. See `process_slashed_deal` for why this matters. + fn unlock_balance_in_table( &mut self, - store: &BS, + locked_table: &mut BalanceTable, addr: &Address, amount: &TokenAmount, lock_reason: Reason, @@ -1111,7 +1137,6 @@ impl State { return Err(actor_error!(illegal_state, "unlock negative amount: {}", amount)); } - let mut locked_table = BalanceTable::from_root(store, &self.locked_table, "locked table")?; locked_table.must_subtract(addr, amount).context("unlocking balance")?; match lock_reason { @@ -1126,14 +1151,33 @@ impl State { } }; - self.locked_table = locked_table.root()?; Ok(()) } - /// move funds from locked in client to available in provider - fn transfer_balance( + fn unlock_balance( &mut self, store: &BS, + addr: &Address, + amount: &TokenAmount, + lock_reason: Reason, + ) -> Result<(), ActorError> + where + BS: Blockstore, + { + let mut locked_table = BalanceTable::from_root(store, &self.locked_table, "locked table")?; + self.unlock_balance_in_table(&mut locked_table, addr, amount, lock_reason)?; + self.locked_table = locked_table.root()?; + Ok(()) + } + + /// move funds from locked in client to available in provider. + /// Takes already-loaded balance tables instead of loading (and re-flushing) them from + /// `self.escrow_table`/`self.locked_table` internally. See `process_slashed_deal` for why + /// this matters. + fn transfer_balance_in_tables( + &mut self, + escrow_table: &mut BalanceTable, + locked_table: &mut BalanceTable, from_addr: &Address, to_addr: &Address, amount: &TokenAmount, @@ -1145,22 +1189,48 @@ impl State { return Err(actor_error!(illegal_state, "transfer negative amount: {}", amount)); } - let mut escrow_table = BalanceTable::from_root(store, &self.escrow_table, "escrow table")?; - // Subtract from locked and escrow tables escrow_table.must_subtract(from_addr, amount)?; - self.unlock_balance(store, from_addr, amount, Reason::ClientStorageFee) + self.unlock_balance_in_table(locked_table, from_addr, amount, Reason::ClientStorageFee) .context("unlocking client balance")?; // Add subtracted amount to the recipient escrow_table.add(to_addr, amount)?; - self.escrow_table = escrow_table.root()?; Ok(()) } - fn slash_balance( + /// move funds from locked in client to available in provider + fn transfer_balance( &mut self, store: &BS, + from_addr: &Address, + to_addr: &Address, + amount: &TokenAmount, + ) -> Result<(), ActorError> + where + BS: Blockstore, + { + let mut escrow_table = BalanceTable::from_root(store, &self.escrow_table, "escrow table")?; + let mut locked_table = BalanceTable::from_root(store, &self.locked_table, "locked table")?; + self.transfer_balance_in_tables( + &mut escrow_table, + &mut locked_table, + from_addr, + to_addr, + amount, + )?; + self.escrow_table = escrow_table.root()?; + self.locked_table = locked_table.root()?; + Ok(()) + } + + /// Takes already-loaded balance tables instead of loading (and re-flushing) them from + /// `self.escrow_table`/`self.locked_table` internally. See `process_slashed_deal` for why + /// this matters. + fn slash_balance_in_tables( + &mut self, + escrow_table: &mut BalanceTable, + locked_table: &mut BalanceTable, addr: &Address, amount: &TokenAmount, lock_reason: Reason, @@ -1172,12 +1242,33 @@ impl State { return Err(actor_error!(illegal_state, "negative amount to slash: {}", amount)); } - let mut escrow_table = BalanceTable::from_root(store, &self.escrow_table, "escrow table")?; - // Subtract from locked and escrow tables escrow_table.must_subtract(addr, amount)?; + self.unlock_balance_in_table(locked_table, addr, amount, lock_reason) + } + + fn slash_balance( + &mut self, + store: &BS, + addr: &Address, + amount: &TokenAmount, + lock_reason: Reason, + ) -> Result<(), ActorError> + where + BS: Blockstore, + { + let mut escrow_table = BalanceTable::from_root(store, &self.escrow_table, "escrow table")?; + let mut locked_table = BalanceTable::from_root(store, &self.locked_table, "locked table")?; + self.slash_balance_in_tables( + &mut escrow_table, + &mut locked_table, + addr, + amount, + lock_reason, + )?; self.escrow_table = escrow_table.root()?; - self.unlock_balance(store, addr, amount, lock_reason) + self.locked_table = locked_table.root()?; + Ok(()) } } diff --git a/actors/miner/src/deadline_state.rs b/actors/miner/src/deadline_state.rs index ba6677a395..495b8f34a1 100644 --- a/actors/miner/src/deadline_state.rs +++ b/actors/miner/src/deadline_state.rs @@ -597,6 +597,7 @@ impl Deadline { Ok((popped, modified)) } + /// Fast path when `record_termination` is false -- see `Partition::terminate_sectors`. #[allow(clippy::too_many_arguments)] pub fn terminate_sectors( &mut self, @@ -607,10 +608,12 @@ impl Deadline { partition_sectors: &mut PartitionSectorMap, sector_size: SectorSize, quant: QuantSpec, - ) -> anyhow::Result { + record_termination: bool, + ) -> anyhow::Result<(PowerPair, Vec)> { let mut partitions = self.partitions_amt(store)?; let mut power_lost = PowerPair::zero(); + let mut terminated_sector_infos = Vec::new(); for (partition_idx, sector_numbers) in partition_sectors.iter() { let mut partition = partitions .get(partition_idx) @@ -622,7 +625,7 @@ impl Deadline { )? .clone(); - let (removed, removed_unproven) = partition + let (removed, removed_unproven, sector_infos) = partition .terminate_sectors( policy, store, @@ -631,6 +634,7 @@ impl Deadline { sector_numbers, sector_size, quant, + record_termination, ) .map_err(|e| { e.downcast_wrap(format!( @@ -644,12 +648,14 @@ impl Deadline { })?; if !removed.is_empty() { - // Record that partition now has pending early terminations. - self.early_terminations.set(partition_idx); + if record_termination { + // Record that partition now has pending early terminations. + self.early_terminations.set(partition_idx); + } // else: fast path -- the caller settles fees itself, so no queue entry is recorded. // Record change to sectors and power self.live_sectors -= removed.len(); - } // note: we should _always_ have early terminations, unless the early termination bitfield is empty. + } self.faulty_power -= &removed.faulty_power; self.live_power -= &removed.active_power; @@ -659,13 +665,15 @@ impl Deadline { // Aggregate power lost from active sectors power_lost += &removed.active_power; + + terminated_sector_infos.extend(sector_infos); } // save partitions back self.partitions = partitions.flush().map_err(|e| e.downcast_wrap("failed to persist partitions"))?; - Ok(power_lost) + Ok((power_lost, terminated_sector_infos)) } /// RemovePartitions removes the specified partitions, shifting the remaining diff --git a/actors/miner/src/lib.rs b/actors/miner/src/lib.rs index c7825f06d9..f1a8143555 100644 --- a/actors/miner/src/lib.rs +++ b/actors/miner/src/lib.rs @@ -2407,92 +2407,127 @@ impl Actor { })?; } - let (had_early_terminations, power_delta) = rt.transaction(|state: &mut State, rt| { - let had_early_terminations = have_pending_early_terminations(state); + let (had_early_terminations, power_delta, sector_size, terminated_sector_infos) = rt + .transaction(|state: &mut State, rt| { + let had_early_terminations = have_pending_early_terminations(state); - let info = get_miner_info(rt.store(), state)?; - - rt.validate_immediate_caller_is( - info.control_addresses.iter().chain(&[info.worker, info.owner]), - )?; + let info = get_miner_info(rt.store(), state)?; - let store = rt.store(); - let curr_epoch = rt.curr_epoch(); - let mut power_delta = PowerPair::zero(); - - let mut deadlines = - state.load_deadlines(store).map_err(|e| e.wrap("failed to load deadlines"))?; + rt.validate_immediate_caller_is( + info.control_addresses.iter().chain(&[info.worker, info.owner]), + )?; - // We're only reading the sectors, so there's no need to save this back. - // However, we still want to avoid re-loading this array per-partition. - let sectors = Sectors::load(store, &state.sectors).map_err(|e| { - e.downcast_default(ExitCode::USR_ILLEGAL_STATE, "failed to load sectors") - })?; + let store = rt.store(); + let curr_epoch = rt.curr_epoch(); + let mut power_delta = PowerPair::zero(); + let mut terminated_sector_infos = Vec::new(); - for (deadline_idx, partition_sectors) in to_process.iter() { - // If the deadline is the current or next deadline to prove, don't allow terminating sectors. - // We assume that deadlines are immutable when being proven. - if !deadline_is_mutable( - rt.policy(), - state.current_proving_period_start(rt.policy(), curr_epoch), - deadline_idx, - curr_epoch, - ) { - return Err(actor_error!( - illegal_argument, - "cannot terminate sectors in immutable deadline {}", - deadline_idx - )); - } + let mut deadlines = + state.load_deadlines(store).map_err(|e| e.wrap("failed to load deadlines"))?; - let quant = state.quant_spec_for_deadline(rt.policy(), deadline_idx); - let mut deadline = deadlines.load_deadline(store, deadline_idx)?; + // We're only reading the sectors, so there's no need to save this back. + // However, we still want to avoid re-loading this array per-partition. + let sectors = Sectors::load(store, &state.sectors).map_err(|e| { + e.downcast_default(ExitCode::USR_ILLEGAL_STATE, "failed to load sectors") + })?; - let removed_power = deadline - .terminate_sectors( + for (deadline_idx, partition_sectors) in to_process.iter() { + // If the deadline is the current or next deadline to prove, don't allow terminating sectors. + // We assume that deadlines are immutable when being proven. + if !deadline_is_mutable( rt.policy(), - store, - §ors, + state.current_proving_period_start(rt.policy(), curr_epoch), + deadline_idx, curr_epoch, - partition_sectors, - info.sector_size, - quant, - ) - .map_err(|e| { - e.downcast_default( - ExitCode::USR_ILLEGAL_STATE, - format!("failed to terminate sectors in deadline {}", deadline_idx), + ) { + return Err(actor_error!( + illegal_argument, + "cannot terminate sectors in immutable deadline {}", + deadline_idx + )); + } + + let quant = state.quant_spec_for_deadline(rt.policy(), deadline_idx); + let mut deadline = deadlines.load_deadline(store, deadline_idx)?; + + // Fast path: settle immediately when there's no backlog (see `settle_terminated_sectors`). + let (removed_power, sector_infos) = deadline + .terminate_sectors( + rt.policy(), + store, + §ors, + curr_epoch, + partition_sectors, + info.sector_size, + quant, + had_early_terminations, ) - })?; + .map_err(|e| { + e.downcast_default( + ExitCode::USR_ILLEGAL_STATE, + format!("failed to terminate sectors in deadline {}", deadline_idx), + ) + })?; - state.early_terminations.set(deadline_idx); - power_delta -= &removed_power; + if had_early_terminations { + state.early_terminations.set(deadline_idx); + } + power_delta -= &removed_power; + terminated_sector_infos.extend(sector_infos); - deadlines.update_deadline(rt.policy(), store, deadline_idx, &deadline).map_err( - |e| { - e.downcast_default( - ExitCode::USR_ILLEGAL_STATE, - format!("failed to update deadline {}", deadline_idx), - ) - }, - )?; - } + deadlines + .update_deadline(rt.policy(), store, deadline_idx, &deadline) + .map_err(|e| { + e.downcast_default( + ExitCode::USR_ILLEGAL_STATE, + format!("failed to update deadline {}", deadline_idx), + ) + })?; + } - state.save_deadlines(store, deadlines).map_err(|e| { - e.downcast_default(ExitCode::USR_ILLEGAL_STATE, "failed to save deadlines") + state.save_deadlines(store, deadlines).map_err(|e| { + e.downcast_default(ExitCode::USR_ILLEGAL_STATE, "failed to save deadlines") + })?; + + Ok((had_early_terminations, power_delta, info.sector_size, terminated_sector_infos)) })?; - Ok((had_early_terminations, power_delta)) - })?; + // Fetched only after the mark transaction above has committed, matching historical + // behavior: an invalid call (bad caller, immutable deadline, ...) fails during marking + // without ever reaching these cross-actor calls. let epoch_reward = request_current_epoch_block_reward(rt)?; let pwr_total = request_current_total_power(rt)?; - // Now, try to process these sectors. - let more = process_early_terminations( - rt, - &epoch_reward.this_epoch_reward_smoothed, - &pwr_total.quality_adj_power_smoothed, - )?; + let more = if had_early_terminations { + // A backlog already existed before this call: fall back to the historical deferred + // path, which will process the queue -- including whatever this call just added to + // it -- in FIFO order. + process_early_terminations( + rt, + &epoch_reward.this_epoch_reward_smoothed, + &pwr_total.quality_adj_power_smoothed, + )? + } else if terminated_sector_infos.is_empty() { + false + } else { + // Fast path: settle immediately using the sector infos already loaded above, in a + // second (much cheaper) transaction that only touches top-level state fields -- + // no re-loading deadlines, partitions, or sector infos. + let settlement = rt.transaction(|state: &mut State, rt| { + settle_terminated_sectors( + state, + rt.store(), + rt.current_balance(), + rt.curr_epoch(), + sector_size, + &epoch_reward.this_epoch_reward_smoothed, + &pwr_total.quality_adj_power_smoothed, + &terminated_sector_infos, + ) + })?; + settlement.apply_side_effects(rt)?; + false + }; if more && !had_early_terminations { // We have remaining terminations, and we didn't _previously_ @@ -4107,6 +4142,149 @@ fn update_existing_sector_info( } // Note: We're using the current power+epoch reward, rather than at time of termination. +/// Accumulates termination-fee inputs for a batch of sectors all terminated at the same epoch. +/// Shared by the immediate-settlement fast path in `terminate_sectors` and the deferred path in +/// `process_early_terminations`, so the fee math can't drift between the two. +#[allow(clippy::too_many_arguments)] +fn accumulate_termination_fees( + sector_infos: &[SectorOnChainInfo], + epoch: ChainEpoch, + sector_size: SectorSize, + reward_smoothed: &FilterEstimate, + quality_adj_power_smoothed: &FilterEstimate, + total_initial_pledge: &mut TokenAmount, + total_penalty: &mut TokenAmount, + terminated_sector_nums: &mut Vec, + sectors_with_data: &mut Vec, +) { + for sector in sector_infos { + *total_initial_pledge += §or.initial_pledge; + let sector_power = qa_power_for_sector(sector_size, sector); + terminated_sector_nums.push(sector.sector_number); + let sector_age = epoch - sector.activation; + let fault_fee = pledge_penalty_for_continued_fault( + reward_smoothed, + quality_adj_power_smoothed, + §or_power, + ); + *total_penalty += + pledge_penalty_for_termination(§or.initial_pledge, sector_age, &fault_fee); + if sector.deal_weight.is_positive() || sector.verified_deal_weight.is_positive() { + sectors_with_data.push(sector.sector_number); + } + } +} + +/// Applies accumulated termination penalty and pledge release to `state`, repaying outstanding +/// fee debt from unlocked pledge. +fn apply_termination_penalty_and_pledge( + state: &mut State, + store: &impl Blockstore, + epoch: ChainEpoch, + balance: &TokenAmount, + total_initial_pledge: TokenAmount, + total_penalty: &TokenAmount, +) -> Result<(TokenAmount, TokenAmount), ActorError> { + // Apply penalty (add to fee debt) + state + .apply_penalty(total_penalty) + .map_err(|e| actor_error!(illegal_state, "failed to apply penalty: {}", e))?; + + // Remove pledge requirement. + let mut pledge_delta = -total_initial_pledge; + state.add_initial_pledge(&pledge_delta).map_err(|e| { + actor_error!(illegal_state, "failed to add initial pledge {}: {}", pledge_delta, e) + })?; + + // Use unlocked pledge to pay down outstanding fee debt + let (penalty, total_unlocked) = state + .repay_partial_debt_in_priority_order(store, epoch, balance) + .map_err(|e| e.downcast_default(ExitCode::USR_ILLEGAL_STATE, "failed to repay penalty"))?; + pledge_delta -= total_unlocked; + + Ok((penalty, pledge_delta)) +} + +/// Result of immediately settling a batch of terminations within the same transaction that +/// marked them, rather than deferring settlement through the early-termination queue. +struct TerminationSettlement { + penalty: TokenAmount, + pledge_delta: TokenAmount, + terminated_sector_nums: Vec, + sectors_with_data: Vec, +} + +impl TerminationSettlement { + fn apply_side_effects(self, rt: &impl Runtime) -> Result<(), ActorError> { + if self.terminated_sector_nums.is_empty() { + return Ok(()); + } + + log::debug!( + "storage provider {} penalized {} for sector termination", + rt.message().receiver(), + self.penalty + ); + burn_funds(rt, self.penalty)?; + notify_pledge_changed(rt, &self.pledge_delta)?; + + let terminated_data = BitField::try_from_bits(self.sectors_with_data) + .context_code(ExitCode::USR_ILLEGAL_STATE, "invalid sector number")?; + request_terminate_deals(rt, rt.curr_epoch(), &terminated_data)?; + + for sector in self.terminated_sector_nums { + emit::sector_terminated(rt, sector)?; + } + + Ok(()) + } +} + +/// Computes and applies (within the caller's transaction) the fee debt, pledge return, and debt +/// repayment for a batch of sectors terminated *right now*, at `epoch`. Only valid when these +/// sectors were never recorded in the early-termination queue (see +/// `Partition::terminate_sectors`'s `record_termination` flag) -- i.e. there's no backlog and +/// nothing else could have been popped alongside them. +#[allow(clippy::too_many_arguments)] +fn settle_terminated_sectors( + state: &mut State, + store: &impl Blockstore, + balance: TokenAmount, + epoch: ChainEpoch, + sector_size: SectorSize, + reward_smoothed: &FilterEstimate, + quality_adj_power_smoothed: &FilterEstimate, + sector_infos: &[SectorOnChainInfo], +) -> Result { + let mut total_initial_pledge = TokenAmount::zero(); + let mut total_penalty = TokenAmount::zero(); + let mut terminated_sector_nums = Vec::with_capacity(sector_infos.len()); + let mut sectors_with_data = Vec::new(); + + accumulate_termination_fees( + sector_infos, + epoch, + sector_size, + reward_smoothed, + quality_adj_power_smoothed, + &mut total_initial_pledge, + &mut total_penalty, + &mut terminated_sector_nums, + &mut sectors_with_data, + ); + + let (penalty, pledge_delta) = apply_termination_penalty_and_pledge( + state, + store, + epoch, + &balance, + total_initial_pledge, + &total_penalty, + )?; + + Ok(TerminationSettlement { penalty, pledge_delta, terminated_sector_nums, sectors_with_data }) +} + fn process_early_terminations( rt: &impl Runtime, reward_smoothed: &FilterEstimate, @@ -4151,47 +4329,27 @@ fn process_early_terminations( .load_sectors(sector_numbers) .map_err(|e| e.wrap("failed to load sector infos"))?; - for sector in §ors { - total_initial_pledge += §or.initial_pledge; - let sector_power = qa_power_for_sector(info.sector_size, sector); - terminated_sector_nums.push(sector.sector_number); - let sector_age = epoch - sector.activation; - let fault_fee = pledge_penalty_for_continued_fault( - reward_smoothed, - quality_adj_power_smoothed, - §or_power, - ); - total_penalty += - pledge_penalty_for_termination(§or.initial_pledge, sector_age, &fault_fee); - if sector.deal_weight.is_positive() || sector.verified_deal_weight.is_positive() { - sectors_with_data.push(sector.sector_number); - } - } + accumulate_termination_fees( + §ors, + epoch, + info.sector_size, + reward_smoothed, + quality_adj_power_smoothed, + &mut total_initial_pledge, + &mut total_penalty, + &mut terminated_sector_nums, + &mut sectors_with_data, + ); } - // Apply penalty (add to fee debt) - state - .apply_penalty(&total_penalty) - .map_err(|e| actor_error!(illegal_state, "failed to apply penalty: {}", e))?; - - // Remove pledge requirement. - let mut pledge_delta = -total_initial_pledge; - state.add_initial_pledge(&pledge_delta).map_err(|e| { - actor_error!(illegal_state, "failed to add initial pledge {}: {}", pledge_delta, e) - })?; - - // Use unlocked pledge to pay down outstanding fee debt - let (penalty, total_unlocked) = state - .repay_partial_debt_in_priority_order( - rt.store(), - rt.curr_epoch(), - &rt.current_balance(), - ) - .map_err(|e| { - e.downcast_default(ExitCode::USR_ILLEGAL_STATE, "failed to repay penalty") - })?; - - pledge_delta -= total_unlocked; + let (penalty, pledge_delta) = apply_termination_penalty_and_pledge( + state, + rt.store(), + rt.curr_epoch(), + &rt.current_balance(), + total_initial_pledge, + &total_penalty, + )?; Ok((result, more, penalty, pledge_delta)) })?; diff --git a/actors/miner/src/partition_state.rs b/actors/miner/src/partition_state.rs index db791e0d09..357a350f18 100644 --- a/actors/miner/src/partition_state.rs +++ b/actors/miner/src/partition_state.rs @@ -478,7 +478,14 @@ impl Partition { /// Marks a collection of sectors as terminated. /// The sectors are removed from Faults and Recoveries. - /// The epoch of termination is recorded for future termination fee calculation. + /// + /// If `record_termination` is true, the epoch of termination is recorded in the partition's + /// early-termination queue for later (deferred) processing, matching historical behavior. + /// If false, the caller takes responsibility for settling the termination (fee calculation, + /// deal notification, etc.) itself, in the same transaction, using the returned sector infos + /// -- this is a fast path available only when there's no pre-existing backlog of deferred + /// terminations, letting the caller skip an otherwise-redundant write-then-immediately-pop + /// round trip through this queue. #[allow(clippy::too_many_arguments)] pub fn terminate_sectors( &mut self, @@ -489,7 +496,8 @@ impl Partition { sector_numbers: &BitField, sector_size: SectorSize, quant: QuantSpec, - ) -> anyhow::Result<(ExpirationSet, PowerPair)> { + record_termination: bool, + ) -> anyhow::Result<(ExpirationSet, PowerPair, Vec)> { let live_sectors = self.live_sectors(); if !live_sectors.contains_all(sector_numbers) { @@ -510,9 +518,11 @@ impl Partition { let removed_sectors = &removed.on_time_sectors | &removed.early_sectors; - // Record early termination. - self.record_early_termination(store, epoch, &removed_sectors) - .map_err(|e| e.downcast_wrap("failed to record early sector termination"))?; + if record_termination { + // Record early termination. + self.record_early_termination(store, epoch, &removed_sectors) + .map_err(|e| e.downcast_wrap("failed to record early sector termination"))?; + } let unproven_nos = &removed_sectors & &self.unproven; @@ -534,7 +544,12 @@ impl Partition { // check invariants self.validate_state()?; - Ok((removed, removed_unproven_power)) + // The caller only needs the terminated sector infos on the fast (non-deferred) path, + // where it settles fees itself using them; on the deferred path they'd just be moved up + // through the deadline and top-level callers and dropped unused. + let terminated_sector_infos = if record_termination { Vec::new() } else { sector_infos }; + + Ok((removed, removed_unproven_power, terminated_sector_infos)) } /// PopExpiredSectors traverses the expiration queue up to and including some epoch, and marks all expiring diff --git a/actors/miner/tests/deadline_state_test.rs b/actors/miner/tests/deadline_state_test.rs index bba6fc8539..0fc727383f 100644 --- a/actors/miner/tests/deadline_state_test.rs +++ b/actors/miner/tests/deadline_state_test.rs @@ -525,7 +525,7 @@ fn terminate_sectors( partition_sector_map.add(partition, sectors).unwrap(); } - deadline.terminate_sectors( + let (power, _) = deadline.terminate_sectors( &Policy::default(), &store, §ors_array, @@ -533,7 +533,9 @@ fn terminate_sectors( &mut partition_sector_map, SECTOR_SIZE, QUANT_SPEC, - ) + true, + )?; + Ok(power) } #[test] diff --git a/actors/miner/tests/miner_actor_test_partitions.rs b/actors/miner/tests/miner_actor_test_partitions.rs index 0f958be919..191b681cdb 100644 --- a/actors/miner/tests/miner_actor_test_partitions.rs +++ b/actors/miner/tests/miner_actor_test_partitions.rs @@ -605,7 +605,7 @@ mod miner_actor_test_partitions { // now terminate 1, 3, 5, and 7 let terminations = make_bitfield(&[1, 3, 5, 7]); let termination_epoch = 3; - let (removed, removed_unproven) = partition + let (removed, removed_unproven, _) = partition .terminate_sectors( &Policy::default(), &rt.store, @@ -614,6 +614,7 @@ mod miner_actor_test_partitions { &terminations, SECTOR_SIZE, QUANT_SPEC, + true, ) .unwrap(); @@ -671,6 +672,7 @@ mod miner_actor_test_partitions { &terminations, SECTOR_SIZE, QUANT_SPEC, + true, ); let err = res.expect_err("expected error, but call succeeded"); @@ -688,7 +690,7 @@ mod miner_actor_test_partitions { let termination_epoch = 3; // First termination works. - let (removed, unproven_power) = partition + let (removed, unproven_power, _) = partition .terminate_sectors( &Policy::default(), &rt.store, @@ -697,6 +699,7 @@ mod miner_actor_test_partitions { &terminations, SECTOR_SIZE, QUANT_SPEC, + true, ) .unwrap(); let expected_active_power = @@ -716,6 +719,7 @@ mod miner_actor_test_partitions { &terminations, SECTOR_SIZE, QUANT_SPEC, + true, ); let err = res.expect_err("expected error, but call succeeded"); @@ -742,6 +746,7 @@ mod miner_actor_test_partitions { &terminations, SECTOR_SIZE, QUANT_SPEC, + true, ) .unwrap(); @@ -942,6 +947,7 @@ mod miner_actor_test_partitions { &terminations, SECTOR_SIZE, QUANT_SPEC, + true, ) .unwrap(); diff --git a/actors/miner/tests/record_skipped_faults.rs b/actors/miner/tests/record_skipped_faults.rs index 438103a4da..a041c0b92f 100644 --- a/actors/miner/tests/record_skipped_faults.rs +++ b/actors/miner/tests/record_skipped_faults.rs @@ -78,6 +78,7 @@ fn already_faulty_and_terminated_sectors_are_ignored() { &terminations, SECTOR_SIZE, QUANT_SPEC, + true, ) .unwrap(); assert_partition_state( diff --git a/test_vm/tests/terminate_gas_bench.rs b/test_vm/tests/terminate_gas_bench.rs new file mode 100644 index 0000000000..bf5cad1e3b --- /dev/null +++ b/test_vm/tests/terminate_gas_bench.rs @@ -0,0 +1,189 @@ +// Ad-hoc benchmark (not part of the regular suite) for measuring blockstore read/write +// activity when a storage provider terminates a batch of sectors, most of them carrying +// verified (DataCap-backed) deals -- the scenario people report as unexpectedly expensive. +// +// Compare before/after the TerminateSectors + OnMinerSectorsTerminate batching fix by +// running this same file against both revisions: +// cargo test -p test_vm --test terminate_gas_bench -- --nocapture + +use fil_actor_market::Method as MarketMethod; +use fil_actor_miner::{Method as MinerMethod, TerminateSectorsParams, TerminationDeclaration}; +use fil_actors_integration_tests::util::{ + advance_by_deadline_to_epoch, advance_to_proving_deadline, create_accounts, create_miner, + cron_tick, make_bitfield, make_piece_manifests_from_deal_ids, market_publish_deal, + miner_prove_sector, precommit_meta_data_from_deals, precommit_sectors_v2, sector_deadline, + submit_windowed_post, verifreg_add_verifier, +}; +use fil_actors_runtime::runtime::Policy; +use fil_actors_runtime::test_blockstores::{BSStats, MemoryBlockstore}; +use fil_actors_runtime::{STORAGE_MARKET_ACTOR_ADDR, VERIFIED_REGISTRY_ACTOR_ADDR}; +use fvm_shared::econ::TokenAmount; +use fvm_shared::piece::PaddedPieceSize; +use fvm_shared::sector::{RegisteredSealProof, StoragePower}; +use num_traits::Zero; +use num_traits::cast::FromPrimitive; +use test_vm::TestVM; +use vm_api::VM; +use vm_api::util::apply_ok; + +/// Number of sectors, each with its own verified deal, that the miner will terminate in one +/// TerminateSectors call. All are sized to fit in a single partition (32GiB partitions hold up +/// to 2349 sectors) so the scenario stresses deal count without needing to seal thousands of +/// sectors just to spill into a second partition. +const NUM_SECTORS: u64 = 1000; + +#[test] +fn terminate_sectors_gas_bench() { + let store = MemoryBlockstore::new(); + let v = TestVM::new_with_singletons(store); + + let addrs = create_accounts(&v, 2, &TokenAmount::from_whole(100_000)); + let (owner, verified_client) = (addrs[0], addrs[1]); + let worker = owner; + + let seal_proof = RegisteredSealProof::StackedDRG32GiBV1P1; + let (miner_id_addr, miner_robust_addr) = create_miner( + &v, + &owner, + &worker, + seal_proof.registered_window_post_proof().unwrap(), + &TokenAmount::from_whole(20 * NUM_SECTORS as i64 + 1_000), + ); + + verifreg_add_verifier(&v, &owner, StoragePower::from_i64(32i64 << 40).unwrap()); + apply_ok( + &v, + &owner, + &VERIFIED_REGISTRY_ACTOR_ADDR, + &TokenAmount::zero(), + fil_actor_verifreg::Method::AddVerifiedClient as u64, + Some(fil_actor_verifreg::VerifierParams { + address: verified_client, + allowance: StoragePower::from_i64(32i64 << 40).unwrap(), + }), + ); + + apply_ok( + &v, + &verified_client, + &STORAGE_MARKET_ACTOR_ADDR, + &TokenAmount::from_whole(3 * NUM_SECTORS as i64), + MarketMethod::AddBalance as u64, + Some(verified_client), + ); + apply_ok( + &v, + &worker, + &STORAGE_MARKET_ACTOR_ADDR, + // 2 FIL provider collateral per deal, plus margin. + &TokenAmount::from_whole(3 * NUM_SECTORS as i64), + MarketMethod::AddBalance as u64, + Some(miner_id_addr), + ); + + let sector_number_base = 100u64; + const DEAL_LIFETIME: i64 = 400 * 2880; // ~400 days + // A verified-deal claim requires term_min(=deal lifetime) <= sector_lifetime <= + // term_max(=deal lifetime + 90-day buffer, capped). Keep comfortably inside that window. + const SECTOR_LIFETIME: i64 = 450 * 2880; // ~450 days + let deal_start = v.epoch() + Policy::default().pre_commit_challenge_delay + 1; + let expiration = deal_start + SECTOR_LIFETIME; + + // Publish one verified deal per sector, then precommit all sectors together in a single + // batch message: sectors precommitted together are packed into the same deadline/partition + // (up to partition capacity -- 2349 sectors for a 32GiB partition), rather than spread across + // deadlines the way one-sector-at-a-time precommits (each its own message, on its own + // schedule) would be. That packing is what lets a real SP's later TerminateSectors call name + // just one (deadline, partition) pair covering many sectors -- and many deals. + let mut deal_ids_by_sector = vec![]; + for i in 0..NUM_SECTORS { + let deals = market_publish_deal( + &v, + &worker, + &verified_client, + &miner_id_addr, + format!("deal{i}"), + PaddedPieceSize(1 << 30), + true, + deal_start, + DEAL_LIFETIME, + ); + deal_ids_by_sector.push(deals.ids.to_vec()); + } + cron_tick(&v); + + let metadata = deal_ids_by_sector + .iter() + .map(|deal_ids| precommit_meta_data_from_deals(&v, deal_ids, seal_proof, false)) + .collect(); + precommit_sectors_v2( + &v, + NUM_SECTORS as usize, + metadata, + &worker, + &miner_robust_addr, + seal_proof, + sector_number_base, + true, + Some(expiration), + ); + + let prove_time = v.epoch() + Policy::default().pre_commit_challenge_delay + 1; + advance_by_deadline_to_epoch(&v, &miner_id_addr, prove_time); + + let mut sector_numbers = vec![]; + for (i, deal_ids) in deal_ids_by_sector.into_iter().enumerate() { + let sector_number = sector_number_base + i as u64; + miner_prove_sector( + &v, + &worker, + &miner_id_addr, + sector_number, + make_piece_manifests_from_deal_ids(&v, deal_ids), + ); + sector_numbers.push(sector_number); + } + + // All sectors should have packed into the same deadline/partition. + let (d_idx, p_idx) = sector_deadline(&v, &miner_id_addr, sector_numbers[0]); + for &s in §or_numbers { + assert_eq!((d_idx, p_idx), sector_deadline(&v, &miner_id_addr, s)); + } + + let (dline_info, p_idx) = advance_to_proving_deadline(&v, &miner_id_addr, sector_numbers[0]); + submit_windowed_post(&v, &worker, &miner_id_addr, dline_info, p_idx, None); + v.set_epoch(dline_info.close); + advance_by_deadline_to_epoch( + &v, + &miner_id_addr, + dline_info.close + Policy::default().deal_updates_interval, + ); + + let before: BSStats = *v.store.stats.borrow(); + + apply_ok( + &v, + &worker, + &miner_robust_addr, + &TokenAmount::zero(), + MinerMethod::TerminateSectors as u64, + Some(TerminateSectorsParams { + terminations: vec![TerminationDeclaration { + deadline: d_idx, + partition: p_idx, + sectors: make_bitfield(§or_numbers), + }], + }), + ); + + let after: BSStats = *v.store.stats.borrow(); + + println!( + "TerminateSectors ({NUM_SECTORS} sectors, {NUM_SECTORS} verified deals): \ + reads +{} writes +{} bytes_read +{} bytes_written +{}", + after.r - before.r, + after.w - before.w, + after.br - before.br, + after.bw - before.bw, + ); +}