Add debt-based trigger for observation sharing

Without this, some entrypoints could get stuck at high
estimates with no way of adjusting back down.

Author: aschran

crates/sui-config/src/node.rs

@@ -220,6 +220,11 @@ pub struct ExecutionTimeObserverConfig {
     /// If unspecified, this will default to `10_000`.
     pub observation_cache_size: Option<NonZeroUsize>,
 
+    /// Size of the channel used for buffering object debt updates from consensus handler.
+    ///
+    /// If unspecified, this will default to `128`.
+    pub object_debt_channel_capacity: Option<NonZeroUsize>,
+
     /// Size of the LRU cache used for tracking object utilization.
     ///
     /// If unspecified, this will default to `50_000`.
@@ -264,6 +269,11 @@ impl ExecutionTimeObserverConfig {
         self.observation_cache_size.unwrap_or(nonzero!(10_000usize))
     }
 
+    pub fn object_debt_channel_capacity(&self) -> NonZeroUsize {
+        self.object_debt_channel_capacity
+            .unwrap_or(nonzero!(128usize))
+    }
+
     pub fn object_utilization_cache_size(&self) -> NonZeroUsize {
         self.object_utilization_cache_size
             .unwrap_or(nonzero!(50_000usize))

crates/sui-core/src/authority/authority_per_epoch_store.rs

@@ -407,6 +407,7 @@ pub struct AuthorityPerEpochStore {
     execution_time_estimator: tokio::sync::Mutex<Option<ExecutionTimeEstimator>>,
     tx_local_execution_time:
         OnceCell<mpsc::Sender<(ProgrammableTransaction, Vec<ExecutionTiming>, Duration)>>,
+    tx_object_debts: OnceCell<mpsc::Sender<Vec<ObjectID>>>,
     // Saved at end of epoch for propagating observations to the next.
     end_of_epoch_execution_time_observations: OnceCell<StoredExecutionTimeObservations>,
 }
@@ -948,6 +949,7 @@ impl AuthorityPerEpochStore {
             randomness_reporter: OnceCell::new(),
             execution_time_estimator: tokio::sync::Mutex::new(execution_time_estimator),
             tx_local_execution_time: OnceCell::new(),
+            tx_object_debts: OnceCell::new(),
             end_of_epoch_execution_time_observations: OnceCell::new(),
         });
 
@@ -1199,19 +1201,23 @@ impl AuthorityPerEpochStore {
         &self.execution_component.executor
     }
 
-    pub fn set_local_execution_time_channel(
+    pub fn set_local_execution_time_channels(
         &self,
         tx_local_execution_time: mpsc::Sender<(
             ProgrammableTransaction,
             Vec<ExecutionTiming>,
             Duration,
         )>,
+        tx_object_debts: mpsc::Sender<Vec<ObjectID>>,
     ) {
         if let Err(e) = self.tx_local_execution_time.set(tx_local_execution_time) {
             debug_fatal!(
                 "failed to set tx_local_execution_time channel on AuthorityPerEpochStore: {e:?}"
             );
         }
+        if let Err(e) = self.tx_object_debts.set(tx_object_debts) {
+            debug_fatal!("failed to set tx_object_debts channel on AuthorityPerEpochStore: {e:?}");
+        }
     }
 
     pub fn record_local_execution_time(
@@ -3566,13 +3572,23 @@ impl AuthorityPerEpochStore {
             .with_label_values(&["randomness_commit"])
             .set(shared_object_using_randomness_congestion_tracker.max_cost() as i64);
 
-        output.set_congestion_control_object_debts(
-            shared_object_congestion_tracker.accumulated_debts(consensus_commit_info),
-        );
-        output.set_congestion_control_randomness_object_debts(
-            shared_object_using_randomness_congestion_tracker
-                .accumulated_debts(consensus_commit_info),
-        );
+        let object_debts =
+            shared_object_congestion_tracker.accumulated_debts(consensus_commit_info);
+        let randomness_object_debts = shared_object_using_randomness_congestion_tracker
+            .accumulated_debts(consensus_commit_info);
+        if let Some(tx_object_debts) = self.tx_object_debts.get() {
+            if let Err(e) = tx_object_debts.try_send(
+                object_debts
+                    .iter()
+                    .map(|(id, _)| *id)
+                    .chain(randomness_object_debts.iter().map(|(id, _)| *id))
+                    .collect(),
+            ) {
+                info!("failed to send updated object debts to ExecutionTimeObserver: {e:?}");
+            }
+        }
+        output.set_congestion_control_object_debts(object_debts);
+        output.set_congestion_control_randomness_object_debts(randomness_object_debts);
 
         if randomness_state_updated {
             if let Some(randomness_manager) = randomness_manager.as_mut() {

crates/sui-core/src/authority/execution_time_estimator.rs

@@ -51,6 +51,9 @@ pub struct ExecutionTimeObserver {
     // via consensus.
     object_utilization_tracker: LruCache<ObjectID, ObjectUtilization>,
 
+    // Sorted list of recently indebted objects, updated by consensus handler.
+    indebted_objects: Vec<ObjectID>,
+
     sharing_rate_limiter: RateLimiter<
         governor::state::NotKeyed,
         governor::state::InMemoryState,
@@ -87,14 +90,17 @@ impl ExecutionTimeObserver {
 
         let (tx_local_execution_time, mut rx_local_execution_time) =
             mpsc::channel(config.observation_channel_capacity().into());
-        epoch_store.set_local_execution_time_channel(tx_local_execution_time);
+        let (tx_object_debts, mut rx_object_debts) =
+            mpsc::channel(config.object_debt_channel_capacity().into());
+        epoch_store.set_local_execution_time_channels(tx_local_execution_time, tx_object_debts);
 
         // TODO: pre-populate local observations with stored data from prior epoch.
         let mut observer = Self {
             epoch_store: Arc::downgrade(&epoch_store),
             consensus_adapter,
             local_observations: LruCache::new(config.observation_cache_size()),
             object_utilization_tracker: LruCache::new(config.object_utilization_cache_size()),
+            indebted_objects: Vec::new(),
             sharing_rate_limiter: RateLimiter::direct(
                 Quota::per_second(config.observation_sharing_rate_limit())
                     .allow_burst(config.observation_sharing_burst_limit()),
@@ -103,10 +109,19 @@ impl ExecutionTimeObserver {
             config,
         };
         spawn_monitored_task!(epoch_store.within_alive_epoch(async move {
-            while let Some((tx, timings, total_duration)) = rx_local_execution_time.recv().await {
-                observer
-                    .record_local_observations(&tx, &timings, total_duration)
-                    .await;
+            loop {
+                tokio::select! {
+                    // TODO: add metrics for messages received.
+                    Some(object_debts) = rx_object_debts.recv() => {
+                        observer.update_indebted_objects(object_debts);
+                    }
+                    Some((tx, timings, total_duration)) = rx_local_execution_time.recv() => {
+                        observer
+                            .record_local_observations(&tx, &timings, total_duration)
+                            .await;
+                    }
+                    else => { break }
+                }
             }
             info!("shutting down ExecutionTimeObserver");
         }));
@@ -136,6 +151,7 @@ impl ExecutionTimeObserver {
             },
             local_observations: LruCache::new(NonZeroUsize::new(10000).unwrap()),
             object_utilization_tracker: LruCache::new(NonZeroUsize::new(50000).unwrap()),
+            indebted_objects: Vec::new(),
             sharing_rate_limiter: RateLimiter::direct(Quota::per_hour(std::num::NonZeroU32::MAX)),
         }
     }
@@ -154,12 +170,19 @@ impl ExecutionTimeObserver {
 
         assert!(tx.commands.len() >= timings.len());
 
+        let mut uses_indebted_object = false;
+
         // Update the accumulated excess execution time for each mutable shared object
         // used in this transaction, and determine the max overage.
         let max_excess_per_object_execution_time = tx
             .shared_input_objects()
             .filter_map(|obj| obj.mutable.then_some(obj.id))
             .map(|id| {
+                // Mark if any object used in the tx is indebted.
+                if !uses_indebted_object && self.indebted_objects.binary_search(&id).is_ok() {
+                    uses_indebted_object = true;
+                }
+
                 // For each object:
                 // - add the execution time of the current transaction to the tracker
                 // - subtract the maximum amount of time available for execution according
@@ -181,11 +204,9 @@ impl ExecutionTimeObserver {
                     utilization.excess_execution_time.saturating_sub(
                         utilization
                             .last_measured
-                            .map(|last_measured| {
-                                now.duration_since(last_measured)
-                                    .mul_f64(self.protocol_params.target_utilization as f64 / 100.0)
-                            })
-                            .unwrap_or(Duration::MAX),
+                            .map(|last_measured| now.duration_since(last_measured))
+                            .unwrap_or(Duration::MAX)
+                            .mul_f64(self.protocol_params.target_utilization as f64 / 100.0),
                     );
                 utilization.last_measured = Some(now);
                 utilization.excess_execution_time
@@ -252,7 +273,7 @@ impl ExecutionTimeObserver {
                 >= self
                     .config
                     .observation_sharing_object_utilization_threshold();
-            if diff_exceeds_threshold && utilization_exceeds_threshold {
+            if diff_exceeds_threshold && (utilization_exceeds_threshold || uses_indebted_object) {
                 debug!("sharing new execution time observation for {key:?}: {new_average:?}");
                 to_share.push((key, new_average));
                 local_observation.last_shared = Some((new_average, Instant::now()));
@@ -297,6 +318,12 @@ impl ExecutionTimeObserver {
             }
         }
     }
+
+    fn update_indebted_objects(&mut self, mut object_debts: Vec<ObjectID>) {
+        object_debts.sort_unstable();
+        object_debts.dedup();
+        self.indebted_objects = object_debts;
+    }
 }
 
 // Key used to save StoredExecutionTimeObservations in the Sui system state object's