Add ObjectBalanceOverdraw benchmark operation and instrumentation (#25227)

## Summary
- Add new `ObjectBalanceOverdraw` operation to composite benchmark
workloads that tests balance pool overdraw scenarios
- Track pool balance in `AccountState` for accurate overdraw
calculations
- Add `assert_reachable!` instrumentation for pending object funds
scenarios to monitor code path coverage

## Test plan
- [ ] Run composite benchmark workloads with the new operation enabled
- [ ] Verify `ObjectBalanceOverdraw` withdraws 50-100% of pool balance
as expected

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

Author: mystenmark

crates/sui-benchmark/src/workloads/composite/mod.rs

@@ -17,6 +17,7 @@ use rand::seq::SliceRandom;
 
 use crate::drivers::Interval;
 use crate::system_state_observer::SystemStateObserver;
+use crate::workloads::composite::operations::ObjectBalanceOverdraw;
 use crate::workloads::payload::{BatchExecutionResults, BatchedTransactionStatus, Payload};
 use crate::workloads::workload::{
     ESTIMATED_COMPUTATION_COST, MAX_GAS_FOR_TESTING, STORAGE_COST_PER_COUNTER, Workload,
@@ -296,6 +297,7 @@ impl CompositeWorkloadConfig {
         probabilities.insert(TestCoinObjectWithdraw::FLAG, 0.1);
         probabilities.insert(AddressBalanceOverdraw::FLAG, 0.1);
         probabilities.insert(AccumulatorBalanceRead::FLAG, 0.3);
+        probabilities.insert(ObjectBalanceOverdraw::FLAG, 0.1);
         Self {
             probabilities,
             alias_tx_probability: 0.3,
@@ -884,7 +886,8 @@ impl Payload for CompositePayload {
         self.current_batch_num_conflicting_transactions = 0;
         let mut transactions = Vec::with_capacity(batch_size + 1);
 
-        let account_state = AccountState::new(sender, &self.fullnode_proxies).await;
+        let account_state =
+            AccountState::new(sender, &self.fullnode_proxies, self.pool.balance_pool).await;
 
         let mut used_gas = vec![];
 
@@ -1052,12 +1055,14 @@ impl Payload for CompositePayload {
 pub struct AccountState {
     pub sender: SuiAddress,
     pub sui_balance: u64,
+    pub pool_balance: u64,
 }
 
 impl AccountState {
     pub async fn new(
         sender: SuiAddress,
         fullnode_proxies: &Vec<Arc<dyn ValidatorProxy + Sync + Send>>,
+        balance_pool: Option<(ObjectID, SequenceNumber)>,
     ) -> Self {
         let mut retries = 0;
         while retries < 3 {
@@ -1069,14 +1074,25 @@ impl AccountState {
                 continue;
             };
             assert_reachable!("successfully got sui balance for address");
+            let pool_balance = if let Some((pool_id, _)) = balance_pool {
+                let pool_address: SuiAddress = pool_id.into();
+                proxy
+                    .get_sui_address_balance(pool_address)
+                    .await
+                    .unwrap_or(0)
+            } else {
+                0
+            };
             return Self {
                 sender,
                 sui_balance,
+                pool_balance,
             };
         }
         Self {
             sender,
             sui_balance: 0,
+            pool_balance: 0,
         }
     }
 }

crates/sui-benchmark/src/workloads/composite/operations.rs

@@ -54,6 +54,7 @@ pub const ALL_OPERATIONS: &[OperationDescriptor] = &[
     TestCoinObjectWithdraw::DESCRIPTOR,
     AddressBalanceOverdraw::DESCRIPTOR,
     AccumulatorBalanceRead::DESCRIPTOR,
+    ObjectBalanceOverdraw::DESCRIPTOR,
 ];
 
 pub fn describe_flags(flags: u32) -> String {
@@ -992,3 +993,89 @@ impl Operation for AccumulatorBalanceRead {
         );
     }
 }
+
+pub struct ObjectBalanceOverdraw;
+
+impl ObjectBalanceOverdraw {
+    pub const FLAG: u32 = 1 << 16;
+    pub const DESCRIPTOR: OperationDescriptor = OperationDescriptor {
+        name: "ObjectBalanceOverdraw",
+        flag: Self::FLAG,
+        factory: || Box::new(ObjectBalanceOverdraw),
+    };
+}
+
+impl Operation for ObjectBalanceOverdraw {
+    fn name(&self) -> &'static str {
+        "object_balance_overdraw"
+    }
+
+    fn operation_flag(&self) -> u32 {
+        Self::FLAG
+    }
+
+    fn resource_requests(&self) -> Vec<ResourceRequest> {
+        vec![ResourceRequest::ObjectBalance]
+    }
+
+    fn init_requirements(&self) -> Vec<InitRequirement> {
+        vec![
+            InitRequirement::CreateBalancePool,
+            InitRequirement::SeedBalancePool,
+        ]
+    }
+
+    fn apply(
+        &self,
+        builder: &mut ProgrammableTransactionBuilder,
+        resources: &OperationResources,
+        account_state: &AccountState,
+    ) {
+        let (pool_id, initial_shared_version) =
+            resources.balance_pool.expect("Balance pool not resolved");
+
+        let withdraw_amount = if account_state.pool_balance == 0 {
+            0
+        } else {
+            let half_balance = std::cmp::max(1, account_state.pool_balance / 2);
+            get_rng().gen_range(half_balance..=account_state.pool_balance)
+        };
+
+        let pool_arg = builder
+            .obj(ObjectArg::SharedObject {
+                id: pool_id,
+                initial_shared_version,
+                mutability: SharedObjectMutability::Mutable,
+            })
+            .unwrap();
+
+        let amount_arg = builder.pure(withdraw_amount).unwrap();
+
+        let withdrawal = builder.programmable_move_call(
+            resources.package_id,
+            Identifier::new("balance_pool").unwrap(),
+            Identifier::new("withdraw").unwrap(),
+            vec![GAS::type_tag()],
+            vec![pool_arg, amount_arg],
+        );
+
+        let balance = builder.programmable_move_call(
+            SUI_FRAMEWORK_PACKAGE_ID,
+            Identifier::new("balance").unwrap(),
+            Identifier::new("redeem_funds").unwrap(),
+            vec![GAS::type_tag()],
+            vec![withdrawal],
+        );
+
+        let coin = builder.programmable_move_call(
+            SUI_FRAMEWORK_PACKAGE_ID,
+            Identifier::new("coin").unwrap(),
+            Identifier::new("from_balance").unwrap(),
+            vec![GAS::type_tag()],
+            vec![balance],
+        );
+
+        let recipient = SuiAddress::random_for_testing_only();
+        builder.transfer_arg(recipient, coin);
+    }
+}

crates/sui-core/src/accumulators/object_funds_checker/mod.rs

@@ -6,6 +6,7 @@ use std::{
     sync::Arc,
 };
 
+use mysten_common::assert_reachable;
 use parking_lot::RwLock;
 use sui_types::{
     SUI_ACCUMULATOR_ROOT_OBJECT_ID,
@@ -132,6 +133,7 @@ impl ObjectFundsChecker {
         match self.check_object_funds(object_withdraws, accumulator_version, funds_read.as_ref()) {
             // Sufficient funds, we can go ahead and commit the execution results as it is.
             ObjectFundsWithdrawStatus::SufficientFunds => {
+                assert_reachable!("object funds sufficient");
                 debug!("Object funds sufficient, committing effects");
                 self.metrics
                     .check_result
@@ -161,13 +163,15 @@ impl ObjectFundsChecker {
                             let tx_digest = cert.digest();
                             match receiver.await {
                                 Ok(FundsWithdrawStatus::MaybeSufficient) => {
+                                    assert_reachable!("object funds maybe sufficient");
                                     // The withdraw state is now deterministically known,
                                     // so we can enqueue the transaction again and it will check again
                                     // whether it is sufficient or not in the next execution.
                                     // TODO: We should be able to optimize this by avoiding re-execution.
                                     debug!(?tx_digest, "Object funds possibly sufficient");
                                 }
                                 Ok(FundsWithdrawStatus::Insufficient) => {
+                                    assert_reachable!("object funds insufficient");
                                     // Re-enqueue with insufficient funds status, so it will be executed
                                     // in the next execution and fail through early error.
                                     // FIXME: We need to also track the amount of gas that was used,

crates/sui-core/src/authority.rs

@@ -1971,6 +1971,7 @@ impl AuthorityState {
                 epoch_store,
             )
         {
+            assert_reachable!("retry object withdraw later");
             return ExecutionOutput::RetryLater;
         }