Instrument code push more (#23536)

Additional instrumentation for code pushes

GitOrigin-RevId: 3cdbbef5f3f753101db4237ef91cdc3d034ec672

crates/application/src/lib.rs

@@ -96,6 +96,7 @@ use database::{
     FastForwardIndexWorker,
     IndexModel,
     IndexWorker,
+    OccRetryStats,
     SearchIndexWorker,
     ShortBoxFuture,
     Snapshot,
@@ -1670,11 +1671,13 @@ impl<RT: Runtime> Application<RT> {
         &self,
         identity: Identity,
         apply_config_args: ApplyConfigArgs,
-    ) -> anyhow::Result<ConfigMetadataAndSchema> {
+    ) -> anyhow::Result<(ConfigMetadataAndSchema, OccRetryStats)> {
         let runner = self.runner.clone();
-        self.execute_with_audit_log_events_and_occ_retries(identity, "apply_config", |tx| {
-            Self::_apply_config(runner.clone(), tx, apply_config_args.clone()).into()
-        })
+        self.execute_with_audit_log_events_and_occ_retries_reporting_stats(
+            identity,
+            "apply_config",
+            |tx| Self::_apply_config(runner.clone(), tx, apply_config_args.clone()).into(),
+        )
         .await
     }
 
@@ -2387,6 +2390,34 @@ impl<RT: Runtime> Application<RT> {
         write_source: impl Into<WriteSource>,
         f: F,
     ) -> anyhow::Result<T>
+    where
+        F: Send + Sync,
+        T: Send + 'static,
+        F: for<'b> Fn(
+            &'b mut Transaction<RT>,
+        ) -> ShortBoxFuture<
+            '_,
+            'a,
+            'b,
+            anyhow::Result<(T, Vec<DeploymentAuditLogEvent>)>,
+        >,
+    {
+        self.execute_with_audit_log_events_and_occ_retries_with_pause_client(
+            identity,
+            PauseClient::new(),
+            write_source,
+            f,
+        )
+        .await
+        .map(|(t, _)| t)
+    }
+
+    pub async fn execute_with_audit_log_events_and_occ_retries_reporting_stats<'a, F, T>(
+        &self,
+        identity: Identity,
+        write_source: impl Into<WriteSource>,
+        f: F,
+    ) -> anyhow::Result<(T, OccRetryStats)>
     where
         F: Send + Sync,
         T: Send + 'static,
@@ -2414,7 +2445,7 @@ impl<RT: Runtime> Application<RT> {
         pause_client: PauseClient,
         write_source: impl Into<WriteSource>,
         f: F,
-    ) -> anyhow::Result<T>
+    ) -> anyhow::Result<(T, OccRetryStats)>
     where
         F: Send + Sync,
         T: Send + 'static,
@@ -2428,7 +2459,7 @@ impl<RT: Runtime> Application<RT> {
         >,
     {
         let db = self.database.clone();
-        let (ts, (t, events)) = db
+        let (ts, (t, events), stats) = db
             .execute_with_occ_retries(
                 identity,
                 FunctionUsageTracker::new(),
@@ -2447,7 +2478,7 @@ impl<RT: Runtime> Application<RT> {
             .try_collect()?;
 
         self.log_sender.send_logs(logs);
-        Ok(t)
+        Ok((t, stats))
     }
 
     pub async fn execute_with_occ_retries<'a, T, F>(
@@ -2466,6 +2497,7 @@ impl<RT: Runtime> Application<RT> {
         self.database
             .execute_with_occ_retries(identity, usage, pause_client, write_source, f)
             .await
+            .map(|(ts, t, _)| (ts, t))
     }
 
     pub async fn shutdown(&self) -> anyhow::Result<()> {

crates/application/src/snapshot_import.rs

@@ -1490,7 +1490,7 @@ async fn finalize_import<RT: Runtime>(
     // If we inserted into an existing table, we're done because the table is
     // now populated and active.
     // If we inserted into an Hidden table, make it Active.
-    let (ts, documents_deleted) = database
+    let (ts, documents_deleted, _) = database
         .execute_with_occ_retries(
             identity,
             FunctionUsageTracker::new(),
@@ -2078,7 +2078,7 @@ async fn prepare_table_for_import<RT: Runtime>(
         insert_into_existing_table_id
     } else {
         let table_number = table_number.or(existing_table_id.map(|id| id.table_number));
-        let (_, table_id) = database
+        let (_, table_id, _) = database
             .execute_with_occ_retries(
                 identity.clone(),
                 FunctionUsageTracker::new(),

crates/database/src/database.rs

@@ -23,7 +23,10 @@ use std::{
         LazyLock,
         OnceLock,
     },
-    time::Duration,
+    time::{
+        Duration,
+        Instant,
+    },
 };
 
 use anyhow::{
@@ -1260,7 +1263,7 @@ impl<RT: Runtime> Database<RT> {
         mut pause_client: PauseClient,
         write_source: impl Into<WriteSource>,
         f: F,
-    ) -> anyhow::Result<(Timestamp, T)>
+    ) -> anyhow::Result<(Timestamp, T, OccRetryStats)>
     where
         T: Send,
         R: Fn(&Error) -> bool,
@@ -1274,6 +1277,7 @@ impl<RT: Runtime> Database<RT> {
                     .begin_with_usage(identity.clone(), usage.clone())
                     .await?;
                 pause_client.wait("retry_tx_loop_start").await;
+                let start = Instant::now();
                 let result = async {
                     let t = f(&mut tx).0.await?;
                     let ts = self
@@ -1282,6 +1286,7 @@ impl<RT: Runtime> Database<RT> {
                     Ok((ts, t))
                 }
                 .await;
+                let duration = Instant::now() - start;
                 match result {
                     Err(e) => {
                         if is_retriable(&e) {
@@ -1294,7 +1299,16 @@ impl<RT: Runtime> Database<RT> {
                             return Err(e);
                         }
                     },
-                    Ok((ts, t)) => return Ok((ts, t)),
+                    Ok((ts, t)) => {
+                        return Ok((
+                            ts,
+                            t,
+                            OccRetryStats {
+                                retries: backoff.failures(),
+                                duration,
+                            },
+                        ))
+                    },
                 }
             }
             let error =
@@ -1312,7 +1326,7 @@ impl<RT: Runtime> Database<RT> {
         pause_client: PauseClient,
         write_source: impl Into<WriteSource>,
         f: F,
-    ) -> anyhow::Result<(Timestamp, T)>
+    ) -> anyhow::Result<(Timestamp, T, OccRetryStats)>
     where
         T: Send,
         F: for<'b> Fn(&'b mut Transaction<RT>) -> ShortBoxFuture<'_, 'a, 'b, anyhow::Result<T>>,
@@ -1878,6 +1892,16 @@ impl<RT: Runtime> Database<RT> {
     }
 }
 
+/// Transaction statistics reported for a retried transaction
+#[derive(Debug, PartialEq, Eq)]
+pub struct OccRetryStats {
+    /// Number of times the transaction was retried. 0 for a transaction that
+    /// succeeded the first time.
+    pub retries: u32,
+    /// The duration of the successful transaction
+    pub duration: Duration,
+}
+
 /// The read that conflicted as part of an OCC
 #[derive(Debug, PartialEq, Eq)]
 pub struct ConflictingRead {

crates/database/src/lib.rs

@@ -136,6 +136,7 @@ pub use self::{
         Database,
         DatabaseSnapshot,
         DocumentDeltas,
+        OccRetryStats,
         ShortBoxFuture,
         ShutdownSignal,
         SnapshotPage,

crates/local_backend/src/deploy_config.rs

@@ -24,6 +24,7 @@ use common::{
     types::NodeDependency,
     version::Version,
 };
+use database::OccRetryStats;
 use errors::{
     ErrorMetadata,
     ErrorMetadataAnyhowExt,
@@ -139,6 +140,42 @@ pub struct ConfigJson {
     pub bundled_module_infos: Option<Vec<BundledModuleInfoJson>>,
 }
 
+static NODE_ENVIRONMENT: &str = "node";
+impl ConfigJson {
+    pub fn stats(&self) -> (usize, usize, usize, usize) {
+        let num_node_modules = self
+            .modules
+            .iter()
+            .filter(|module| module.environment.as_deref() == Some(NODE_ENVIRONMENT))
+            .count();
+        let size_node_modules = self
+            .modules
+            .iter()
+            .filter(|module| module.environment.as_deref() == Some(NODE_ENVIRONMENT))
+            .fold(0, |acc, e| {
+                acc + e.source.len() + e.source_map.as_ref().map_or(0, |sm| sm.len())
+            });
+        let size_v8_modules = self
+            .modules
+            .iter()
+            .filter(|module| module.environment.as_deref() != Some(NODE_ENVIRONMENT))
+            .fold(0, |acc, e| {
+                acc + e.source.len() + e.source_map.as_ref().map_or(0, |sm| sm.len())
+            });
+        let num_v8_modules = self
+            .modules
+            .iter()
+            .filter(|module| module.environment.as_deref() != Some(NODE_ENVIRONMENT))
+            .count();
+        (
+            num_v8_modules,
+            num_node_modules,
+            size_v8_modules,
+            size_node_modules,
+        )
+    }
+}
+
 /// API level structure for representing modules as Json
 #[derive(Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
@@ -198,6 +235,7 @@ pub struct PushAnalytics {
     pub udf_server_version: Version,
     pub analyze_results: BTreeMap<CanonicalizedModulePath, AnalyzedModule>,
     pub schema: Option<DatabaseSchema>,
+    pub occ_stats: OccRetryStats,
 }
 
 #[debug_handler]
@@ -298,10 +336,13 @@ pub async fn push_config_handler(
         source_package.clone(),
     )
     .await?;
-    let ConfigMetadataAndSchema {
-        config_metadata,
-        schema,
-    } = application
+    let (
+        ConfigMetadataAndSchema {
+            config_metadata,
+            schema,
+        },
+        occ_stats,
+    ) = application
         .apply_config_with_retries(
             identity.clone(),
             ApplyConfigArgs {
@@ -324,6 +365,7 @@ pub async fn push_config_handler(
             udf_server_version: udf_config.server_version,
             analyze_results: analyze_result,
             schema,
+            occ_stats,
         },
     ))
 }