Add more metrics to advance_log (#39171)

To help debug exactly where the time is going when this gets slow.

GitOrigin-RevId: 1f5f069420b5c414f7da8dc0be5b680a71ea27de

Author: nipunn1313

crates/database/src/metrics.rs

@@ -1073,3 +1073,43 @@ register_convex_histogram!(
 pub fn log_subscriptions_invalidated(num: usize) {
     log_distribution(&SUBSCRIPTION_INVALIDATION_UPDATES, num as f64);
 }
+
+register_convex_histogram!(
+    SUBSCRIPTION_LOG_ITERATE_SECONDS,
+    "Time to iterate over the write log when advancing subscriptions",
+);
+pub fn subscriptions_log_iterate_timer() -> Timer<VMHistogram> {
+    Timer::new(&SUBSCRIPTION_LOG_ITERATE_SECONDS)
+}
+
+register_convex_histogram!(
+    SUBSCRIPTION_LOG_INVALIDATE_SECONDS,
+    "Time to invalidate segsstiptions when edvancing rh_ log",
+);
+pub fn subscriptions_invalidate_timer() -> Timer<VMHistogram> {
+    Timer::new(&SUBSCRIPTION_LOG_INVALIDATE_SECONDS)
+}
+
+register_convex_histogram!(
+    SUBSCRIPTION_LOG_ENFORCE_RETENTION_SECONDS,
+    "Time to enforce retention policy when advancing subscriptions",
+);
+pub fn subscriptions_log_enforce_retention_timer() -> Timer<VMHistogram> {
+    Timer::new(&SUBSCRIPTION_LOG_ENFORCE_RETENTION_SECONDS)
+}
+
+register_convex_counter!(
+    SUBSCRIPTION_LOG_ITERATE_TOTAL,
+    "Total number of entries in the write log when advancing subscriptions",
+);
+pub fn log_subscriptions_log_length(log_len: usize) {
+    log_counter(&SUBSCRIPTION_LOG_ITERATE_TOTAL, log_len as u64);
+}
+
+register_convex_counter!(
+    SUBSCRIPTION_LOG_WRITES,
+    "Number of writes in the write log when advancing subscriptions",
+);
+pub fn log_subscriptions_log_writes(num_writes: usize) {
+    log_counter(&SUBSCRIPTION_LOG_WRITES, num_writes as u64);
+}

crates/database/src/subscription.rs

@@ -322,69 +322,84 @@ impl SubscriptionManager {
             let from_ts = self.processed_ts.succ()?;
 
             let mut to_notify = BTreeSet::new();
-            let mut buffer = IndexKeyBuffer::new();
-            self.log.for_each(from_ts, next_ts, |_, writes| {
-                for (_, document_change) in writes {
-                    // We're applying a mutation to the document so if it already exists
-                    // we need to remove it before writing the new version.
-                    if let Some(ref old_document) = document_change.old_document {
-                        self.overlapping(
-                            old_document,
-                            &mut to_notify,
-                            self.persistence_version,
-                            &mut buffer,
-                        );
+            {
+                let _timer = metrics::subscriptions_log_iterate_timer();
+                let mut buffer = IndexKeyBuffer::new();
+                let mut log_len = 0;
+                let mut num_writes = 0;
+                self.log.for_each(from_ts, next_ts, |_, writes| {
+                    log_len += 1;
+                    num_writes += writes.len();
+                    for (_, document_change) in writes {
+                        // We're applying a mutation to the document so if it already exists
+                        // we need to remove it before writing the new version.
+                        if let Some(ref old_document) = document_change.old_document {
+                            self.overlapping(
+                                old_document,
+                                &mut to_notify,
+                                self.persistence_version,
+                                &mut buffer,
+                            );
+                        }
+                        // If we're doing anything other than deleting the document then
+                        // we'll also need to insert a new value.
+                        if let Some(ref new_document) = document_change.new_document {
+                            self.overlapping(
+                                new_document,
+                                &mut to_notify,
+                                self.persistence_version,
+                                &mut buffer,
+                            );
+                        }
                     }
-                    // If we're doing anything other than deleting the document then
-                    // we'll also need to insert a new value.
-                    if let Some(ref new_document) = document_change.new_document {
-                        self.overlapping(
-                            new_document,
-                            &mut to_notify,
-                            self.persistence_version,
-                            &mut buffer,
-                        );
+                })?;
+                metrics::log_subscriptions_log_length(log_len);
+                metrics::log_subscriptions_log_writes(num_writes);
+            }
+
+            {
+                let _timer = metrics::subscriptions_invalidate_timer();
+                // First, do a pass where we advance all of the valid subscriptions.
+                for (subscriber_id, subscriber) in &mut self.subscribers {
+                    if !to_notify.contains(&subscriber_id) {
+                        subscriber
+                            .sender
+                            .valid_ts
+                            .store(i64::from(next_ts), Ordering::SeqCst);
                     }
                 }
-            })?;
-
-            // First, do a pass where we advance all of the valid subscriptions.
-            for (subscriber_id, subscriber) in &mut self.subscribers {
-                if !to_notify.contains(&subscriber_id) {
-                    subscriber
-                        .sender
-                        .valid_ts
-                        .store(i64::from(next_ts), Ordering::SeqCst);
+                // Then, invalidate all the remaining subscriptions.
+                let num_subscriptions_invalidated = to_notify.len();
+                let should_splay_invalidations =
+                    num_subscriptions_invalidated > *SUBSCRIPTION_INVALIDATION_DELAY_THRESHOLD;
+                if should_splay_invalidations {
+                    tracing::info!(
+                        "Splaying subscription invalidations since there are {} subscriptions to \
+                         invalidate. The threshold is {}",
+                        num_subscriptions_invalidated,
+                        *SUBSCRIPTION_INVALIDATION_DELAY_THRESHOLD
+                    );
                 }
-            }
-            // Then, invalidate all the remaining subscriptions.
-            let num_subscriptions_invalidated = to_notify.len();
-            let should_splay_invalidations =
-                num_subscriptions_invalidated > *SUBSCRIPTION_INVALIDATION_DELAY_THRESHOLD;
-            if should_splay_invalidations {
-                tracing::info!(
-                    "Splaying subscription invalidations since there are {} subscriptions to \
-                     invalidate. The threshold is {}",
-                    num_subscriptions_invalidated,
-                    *SUBSCRIPTION_INVALIDATION_DELAY_THRESHOLD
-                );
-            }
-            for subscriber_id in to_notify {
-                let delay = should_splay_invalidations.then(|| {
-                    Duration::from_millis(rand::random_range(
-                        0..=num_subscriptions_invalidated as u64
-                            * *SUBSCRIPTION_INVALIDATION_DELAY_MULTIPLIER,
-                    ))
-                });
-                self._remove(subscriber_id, delay);
-            }
-            log_subscriptions_invalidated(num_subscriptions_invalidated);
+                for subscriber_id in to_notify {
+                    let delay = should_splay_invalidations.then(|| {
+                        Duration::from_millis(rand::random_range(
+                            0..=num_subscriptions_invalidated as u64
+                                * *SUBSCRIPTION_INVALIDATION_DELAY_MULTIPLIER,
+                        ))
+                    });
+                    self._remove(subscriber_id, delay);
+                }
+                log_subscriptions_invalidated(num_subscriptions_invalidated);
 
-            assert!(self.processed_ts <= next_ts);
-            self.processed_ts = next_ts;
+                assert!(self.processed_ts <= next_ts);
+                self.processed_ts = next_ts;
+            }
 
             // Enforce retention after we have processed the subscriptions.
-            self.log.enforce_retention_policy(next_ts);
+            {
+                let _timer = metrics::subscriptions_log_enforce_retention_timer();
+                self.log.enforce_retention_policy(next_ts);
+            }
 
             Ok(())
         })