Open source the system cleanup worker (#30519)

1. Good for cleaning up the session metadata for mutations
2. Will be good to have when we are using it for export expiration
   which feels more of a product feature.

GitOrigin-RevId: 6c5630c765bac8b8915bf6aa1b1644a2d2ce2655

crates/application/Cargo.toml

@@ -28,6 +28,7 @@ float_next_after = { workspace = true }
 function_runner = { path = "../function_runner" }
 futures = { workspace = true }
 futures-async-stream = { workspace = true }
+governor = { workspace = true }
 headers = { workspace = true }
 http = { workspace = true }
 http_client = { path = "../../crates/http_client" }

crates/application/src/lib.rs

@@ -278,6 +278,7 @@ use sync_types::{
     ModulePath,
     SerializedQueryJournal,
 };
+use system_table_cleanup::SystemTableCleanupWorker;
 use table_summary_worker::{
     TableSummaryClient,
     TableSummaryWorker,
@@ -336,6 +337,7 @@ pub mod redaction;
 pub mod scheduled_jobs;
 mod schema_worker;
 pub mod snapshot_import;
+mod system_table_cleanup;
 mod table_summary_worker;
 pub mod valid_identifier;
 
@@ -482,6 +484,7 @@ pub struct Application<RT: Runtime> {
     schema_worker: Arc<Mutex<Box<dyn SpawnHandle>>>,
     snapshot_import_worker: Arc<Mutex<Box<dyn SpawnHandle>>>,
     export_worker: Arc<Mutex<Box<dyn SpawnHandle>>>,
+    system_table_cleanup_worker: Arc<Mutex<Box<dyn SpawnHandle>>>,
     log_sender: Arc<dyn LogSender>,
     log_visibility: Arc<dyn LogVisibility<RT>>,
     module_cache: ModuleCache<RT>,
@@ -515,6 +518,7 @@ impl<RT: Runtime> Clone for Application<RT> {
             schema_worker: self.schema_worker.clone(),
             snapshot_import_worker: self.snapshot_import_worker.clone(),
             export_worker: self.export_worker.clone(),
+            system_table_cleanup_worker: self.system_table_cleanup_worker.clone(),
             log_sender: self.log_sender.clone(),
             log_visibility: self.log_visibility.clone(),
             module_cache: self.module_cache.clone(),
@@ -591,6 +595,12 @@ impl<RT: Runtime> Application<RT> {
             SchemaWorker::start(runtime.clone(), database.clone()),
         )));
 
+        let system_table_cleanup_worker =
+            SystemTableCleanupWorker::new(runtime.clone(), database.clone());
+        let system_table_cleanup_worker = Arc::new(Mutex::new(
+            runtime.spawn("system_table_cleanup_worker", system_table_cleanup_worker),
+        ));
+
         let function_log = FunctionExecutionLog::new(
             runtime.clone(),
             database.usage_counter(),
@@ -678,6 +688,7 @@ impl<RT: Runtime> Application<RT> {
             schema_worker,
             export_worker,
             snapshot_import_worker,
+            system_table_cleanup_worker,
             log_sender,
             log_visibility,
             module_cache,

crates/application/src/system_table_cleanup/metrics.rs

@@ -0,0 +1,30 @@
+use metrics::{
+    log_counter_with_labels,
+    prometheus::VMHistogram,
+    register_convex_counter,
+    register_convex_histogram,
+    StaticMetricLabel,
+    Timer,
+};
+use value::TableName;
+
+register_convex_histogram!(
+    SYSTEM_TABLE_CLEANUP_SECONDS,
+    "Duration of system table cleanup"
+);
+pub fn system_table_cleanup_timer() -> Timer<VMHistogram> {
+    Timer::new(&SYSTEM_TABLE_CLEANUP_SECONDS)
+}
+
+register_convex_counter!(
+    SYSTEM_TABLE_CLEANUP_ROWS_TOTAL,
+    "Number of rows cleaned up in system tables",
+    &["table"]
+);
+pub fn log_system_table_cleanup_rows(table_name: &TableName, rows: usize) {
+    log_counter_with_labels(
+        &SYSTEM_TABLE_CLEANUP_ROWS_TOTAL,
+        rows as u64,
+        vec![StaticMetricLabel::new("table", table_name.to_string())],
+    )
+}