Split DATABASE_WORKERS_MAX_CHECKPOINT_AGE into search and table-summary knobs (#42873)

GitOrigin-RevId: c3e2b97e406cd365e717524150cea40a15157364

Author: goffrie

crates/application/src/table_summary_worker.rs

@@ -10,10 +10,10 @@ use common::{
         LeaseLostError,
     },
     knobs::{
-        DATABASE_WORKERS_MAX_CHECKPOINT_AGE,
         DATABASE_WORKERS_MIN_COMMITS,
         TABLE_SUMMARY_AGE_JITTER_SECONDS,
         TABLE_SUMMARY_BOOTSTRAP_RECENT_THRESHOLD,
+        TABLE_SUMMARY_MAX_CHECKPOINT_AGE,
     },
     persistence::Persistence,
     runtime::{
@@ -136,12 +136,10 @@ impl<RT: Runtime> TableSummaryWorker<RT> {
 
     fn jittered_max_age(&self) -> Duration {
         let max_age_jitter = (*TABLE_SUMMARY_AGE_JITTER_SECONDS)
-            .min(DATABASE_WORKERS_MAX_CHECKPOINT_AGE.as_secs_f32() / 2.0)
+            .min(TABLE_SUMMARY_MAX_CHECKPOINT_AGE.as_secs_f32() / 2.0)
             * self.runtime.rng().random_range(-1.0..=1.0);
-        Duration::try_from_secs_f32(
-            DATABASE_WORKERS_MAX_CHECKPOINT_AGE.as_secs_f32() + max_age_jitter,
-        )
-        .unwrap_or_default()
+        Duration::try_from_secs_f32(TABLE_SUMMARY_MAX_CHECKPOINT_AGE.as_secs_f32() + max_age_jitter)
+            .unwrap_or_default()
     }
 
     async fn go(self, cancel_receiver: oneshot::Receiver<()>, lease_lost_shutdown: ShutdownSignal) {

crates/common/src/knobs.rs

@@ -524,7 +524,7 @@ pub static SEARCH_INDEX_WORKER_PAGES_PER_SECOND: LazyLock<NonZeroU32> = LazyLock
     )
 });
 
-/// Don't allow database workers to have more than an hour of uncheckpointed
+/// Don't allow search index workers to have more than an hour of uncheckpointed
 /// data.
 ///
 /// For search/vector index workers - Note that fast-forwarding will keep the
@@ -535,8 +535,8 @@ pub static SEARCH_INDEX_WORKER_PAGES_PER_SECOND: LazyLock<NonZeroU32> = LazyLock
 /// DocumentRevisionStream to build new segments, so this value needs to be low
 /// enough to not block the search index flushers for too long, or else writes
 /// will start failing. This is why we set this value lower for pro users (10m).
-pub static DATABASE_WORKERS_MAX_CHECKPOINT_AGE: LazyLock<Duration> =
-    LazyLock::new(|| Duration::from_secs(env_config("DATABASE_WORKERS_MAX_CHECKPOINT_AGE", 3600)));
+pub static SEARCH_WORKERS_MAX_CHECKPOINT_AGE: LazyLock<Duration> =
+    LazyLock::new(|| Duration::from_secs(env_config("SEARCH_WORKERS_MAX_CHECKPOINT_AGE", 3600)));
 
 /// Don't fast-forward an index less than ten seconds forward so we don't
 /// amplify every commit into another write when the system is under heavy load.
@@ -1237,19 +1237,24 @@ pub static MAX_SEARCHLIGHT_REQUEST_SIZE: LazyLock<usize> =
 /// other than the other workers happens.
 ///
 /// We must also ensure that workers advance periodically to ensure that we can
-/// run document retention in the future. Database times are bumped periodically
+/// run document retention. Database times are bumped periodically
 /// even if no writes occur. So any worker that checks this should always have
 /// some maximum period of time after which they checkpoint unconditionally.
 pub static DATABASE_WORKERS_MIN_COMMITS: LazyLock<usize> =
     LazyLock::new(|| env_config("DATABASE_WORKERS_MIN_COMMITS", 500));
 
+/// Update table summaries for idle instances once every 4 hours.
+pub static TABLE_SUMMARY_MAX_CHECKPOINT_AGE: LazyLock<Duration> = LazyLock::new(|| {
+    Duration::from_secs(env_config("TABLE_SUMMARY_MAX_CHECKPOINT_AGE", 4 * 60 * 60))
+});
+
 /// The TableSummaryWorker must checkpoint every
-/// [`DATABASE_WORKERS_MAX_CHECKPOINT_AGE`] seconds even if nothing has changed.
+/// [`TABLE_SUMMARY_MAX_CHECKPOINT_AGE`] seconds even if nothing has changed.
 /// However, to prevent all instances from checkpointing at the same time, we'll
 /// add a jitter of up to ±TABLE_SUMMARY_AGE_JITTER_SECONDS.
 ///
 /// Note: the configured value is capped at
-/// `DATABASE_WORKERS_MAX_CHECKPOINT_AGE/2`.
+/// `TABLE_SUMMARY_MAX_CHECKPOINT_AGE/2`.
 pub static TABLE_SUMMARY_AGE_JITTER_SECONDS: LazyLock<f32> =
     LazyLock::new(|| env_config("TABLE_SUMMARY_AGE_JITTER_SECONDS", 900.0));
 

crates/database/src/search_index_workers/fast_forward.rs

@@ -20,10 +20,10 @@ use common::{
         Interval,
     },
     knobs::{
-        DATABASE_WORKERS_MAX_CHECKPOINT_AGE,
         DATABASE_WORKERS_MIN_COMMITS,
         DATABASE_WORKERS_POLL_INTERVAL,
         INDEX_WORKERS_INITIAL_BACKOFF,
+        SEARCH_WORKERS_MAX_CHECKPOINT_AGE,
     },
     persistence::PersistenceSnapshot,
     query::Order,
@@ -177,7 +177,7 @@ impl FastForwardIndexWorker {
         if let Some(last_fast_forward_info) = last_fast_forward_info
             && commits_since_load - last_fast_forward_info.observed_commits
                 < *DATABASE_WORKERS_MIN_COMMITS
-            && now - last_fast_forward_info.ts < *DATABASE_WORKERS_MAX_CHECKPOINT_AGE
+            && now - last_fast_forward_info.ts < *SEARCH_WORKERS_MAX_CHECKPOINT_AGE
         {
             tracing::debug!(
                 "{log_name} not enough commits and too recent to fast forward: {}, {:?}",

crates/database/src/search_index_workers/search_flusher.rs

@@ -16,8 +16,8 @@ use anyhow::Context;
 use common::{
     bootstrap_model::index::IndexMetadata,
     knobs::{
-        DATABASE_WORKERS_MAX_CHECKPOINT_AGE,
         DEFAULT_DOCUMENTS_PAGE_SIZE,
+        SEARCH_WORKERS_MAX_CHECKPOINT_AGE,
         VECTOR_INDEX_WORKER_PAGE_SIZE,
     },
     persistence::{
@@ -297,7 +297,7 @@ impl<RT: Runtime, T: SearchIndex + 'static> SearchFlusher<RT, T> {
                     anyhow::ensure!(ts <= *step_ts);
 
                     let index_age = *step_ts - ts;
-                    let too_old = (index_age >= *DATABASE_WORKERS_MAX_CHECKPOINT_AGE
+                    let too_old = (index_age >= *SEARCH_WORKERS_MAX_CHECKPOINT_AGE
                         && index_size > 0)
                         .then_some(BuildReason::TooOld);
                     if too_old.is_some() {

crates/database/src/tests/randomized_search_tests.rs

@@ -21,7 +21,7 @@ use common::{
         IndexMetadata,
     },
     floating_point::assert_approx_equal,
-    knobs::DATABASE_WORKERS_MAX_CHECKPOINT_AGE,
+    knobs::SEARCH_WORKERS_MAX_CHECKPOINT_AGE,
     pause::PauseController,
     persistence::Persistence,
     query::{
@@ -1435,7 +1435,7 @@ async fn test_flushing_does_not_invalidate_subscriptions(rt: TestRuntime) -> any
     // Force a checkpoint by advancing time
     scenario
         .rt
-        .advance_time(*DATABASE_WORKERS_MAX_CHECKPOINT_AGE * 2)
+        .advance_time(*SEARCH_WORKERS_MAX_CHECKPOINT_AGE * 2)
         .await;
     scenario.insert("new text", "b").await?;
     scenario.backfill().await?;

crates/database/src/text_index_worker/fast_forward.rs

@@ -171,7 +171,7 @@ pub mod tests {
         );
 
         // Check that we fast-forward if we advance time sufficiently far forward past
-        // DATABASE_WORKERS_MAX_CHECKPOINT_AGE even with no writes.
+        // SEARCH_WORKERS_MAX_CHECKPOINT_AGE even with no writes.
         rt.advance_time(Duration::from_secs(7200)).await;
         database.bump_max_repeatable_ts().await?;
         let metrics = fast_forward(&rt, database, &mut last_fast_forward_info).await?;

crates/database/src/vector_index_worker/fast_forward.rs

@@ -176,7 +176,7 @@ mod tests {
         );
 
         // Check that we fast-forward if we advance time sufficiently far forward past
-        // DATABASE_WORKERS_MAX_CHECKPOINT_AGE.
+        // SEARCH_WORKERS_MAX_CHECKPOINT_AGE.
         rt.advance_time(Duration::from_secs(7200)).await;
         database.bump_max_repeatable_ts().await?;
         let metrics = fast_forward(&rt, &database, &mut last_fast_forward_info).await?;