Reorder and retime some startup logs so it's clearer (#25581)

Goal - final logs are listening on port from just run-local-backend

```
2024-05-10T21:41:36.646131Z  INFO common::http: Listening on
http://0.0.0.0:8000
2024-05-10T21:41:36.646370Z  INFO local_backend::proxy: Starting dev
site proxy at 0.0.0.0:3211...
2024-05-10T21:41:36.646491Z  INFO common::http: Listening on
http://0.0.0.0:3211
```
GitOrigin-RevId: 02b2f23e77193db18b1cd4640bea3f8879a3979c

Author: nipunn1313

crates/application/src/table_summary_worker.rs

@@ -91,7 +91,7 @@ impl<RT: Runtime> TableSummaryWorker<RT> {
         {
             return Ok(());
         }
-        tracing::info!("Writing table summary");
+        tracing::info!("Writing table summary checkpoint");
         let snapshot = writer.compute_from_last_checkpoint().await?;
         write_snapshot(self.persistence.as_ref(), &snapshot).await?;
         *last_write_info = Some(LastWriteInfo {
@@ -115,22 +115,20 @@ impl<RT: Runtime> TableSummaryWorker<RT> {
 
         let mut last_write_info = None;
         loop {
+            let result = self
+                .checkpoint_table_summaries(&mut last_write_info, &writer)
+                .await;
+            if let Err(mut err) = result {
+                report_error(&mut err);
+            }
             let wait_fut = self.runtime.wait(Duration::from_secs(10)).fuse();
             pin_mut!(wait_fut);
             select_biased! {
                 _ = cancel_fut => {
                     tracing::info!("Shutting down table summary worker...");
                     break;
                 }
-                _ = wait_fut => {
-                    let result = self.checkpoint_table_summaries(
-                        &mut last_write_info,
-                        &writer,
-                    ).await;
-                    if let Err(mut err) = result {
-                        report_error(&mut err);
-                    }
-                },
+                _ = wait_fut => {},
             }
         }
     }

crates/database/src/index_workers/fast_forward.rs

@@ -2,6 +2,7 @@ use std::{
     cmp::max,
     collections::BTreeSet,
     future::Future,
+    time::Duration,
 };
 
 use async_trait::async_trait;
@@ -95,7 +96,13 @@ impl FastForwardIndexWorker {
         rt: RT,
         db: Database<RT>,
     ) -> impl Future<Output = ()> + Send {
-        retry_loop_expect_occs_and_overloaded("FastForwardWorker", rt, db, FastForwardIndexWorker)
+        retry_loop_expect_occs_and_overloaded(
+            "FastForwardWorker",
+            rt,
+            db,
+            Duration::ZERO,
+            FastForwardIndexWorker,
+        )
     }
 
     async fn fast_forward_loop<RT: Runtime>(

crates/database/src/index_workers/retriable_worker.rs

@@ -1,3 +1,6 @@
+use std::time::Duration;
+
+use anyhow::Context;
 use async_trait::async_trait;
 use common::{
     errors::report_error,
@@ -34,8 +37,11 @@ pub(crate) async fn retry_loop_expect_occs_and_overloaded<RT: Runtime>(
     name: &'static str,
     runtime: RT,
     db: Database<RT>,
+    initial_wait: Duration,
     work: impl RetriableWorker<RT>,
 ) {
+    tracing::info!("Starting {name}");
+    runtime.wait(initial_wait).await;
     retry_failures(name, runtime, db, MAX_OVERLOADED_RETRIES, work).await
 }
 
@@ -49,9 +55,12 @@ async fn retry_failures<RT: Runtime>(
     let mut backoff = Backoff::new(*INDEX_WORKERS_INITIAL_BACKOFF, MAX_BACKOFF);
     let mut occ_errors = 0;
     let mut overloaded_errors = 0;
-
     loop {
-        if let Err(e) = work.work_loop(name, &runtime, &db, &mut backoff).await {
+        if let Err(mut e) = work
+            .work_loop(name, &runtime, &db, &mut backoff)
+            .await
+            .context(format!("{name} died"))
+        {
             if e.is_occ() {
                 occ_errors += 1;
                 // Do not reset overloaded errors because we expect
@@ -73,11 +82,11 @@ async fn retry_failures<RT: Runtime>(
 
             let expected_error = expected_occ || expected_overloaded;
             if !expected_error {
-                report_error(&mut e.context(format!("{name} died")));
+                report_error(&mut e);
             }
             let delay = runtime.with_rng(|rng| backoff.fail(rng));
             tracing::error!(
-                "{name} died, num_failures: {}. Backing off for {}ms, expected: {}",
+                "{name} died, num_failures: {}. Backing off for {}ms, expected: {}: {e:#}",
                 backoff.failures(),
                 delay.as_millis(),
                 expected_error,

crates/database/src/index_workers/search_worker.rs

@@ -1,4 +1,7 @@
-use std::sync::Arc;
+use std::{
+    sync::Arc,
+    time::Duration,
+};
 
 use async_trait::async_trait;
 use common::{
@@ -67,6 +70,8 @@ impl<RT: Runtime> SearchIndexWorker<RT> {
             "VectorFlusher",
             runtime.clone(),
             database.clone(),
+            // Wait a bit since vector needs time to bootstrap. Makes startup logs a bit cleaner.
+            Duration::from_secs(5),
             SearchIndexWorker::VectorFlusher(VectorIndexFlusher::new(
                 runtime.clone(),
                 database.clone(),
@@ -78,6 +83,7 @@ impl<RT: Runtime> SearchIndexWorker<RT> {
             "VectorCompactor",
             runtime.clone(),
             database.clone(),
+            Duration::ZERO,
             SearchIndexWorker::VectorCompactor(VectorIndexCompactor::new(
                 database.clone(),
                 searcher,
@@ -90,6 +96,7 @@ impl<RT: Runtime> SearchIndexWorker<RT> {
             "SearchFlusher",
             runtime.clone(),
             database.clone(),
+            Duration::ZERO,
             SearchIndexWorker::SearchFlusher(SearchIndexFlusher::new(
                 runtime,
                 database.clone(),
@@ -107,8 +114,6 @@ impl<RT: Runtime> SearchIndexWorker<RT> {
         db: &Database<RT>,
         backoff: &mut Backoff,
     ) -> anyhow::Result<()> {
-        tracing::info!("Starting {name}");
-
         loop {
             let status = log_worker_starting(name);
             let (metrics, token) = match self {