Clean up after migration BackfillIndexState migration (#24073)

Clean up old migrations + make index_state_lower_bound required.

GitOrigin-RevId: 07f255858a32c9919748eb019f6cff96f6e58ae2

Author: Preslav Le

crates/common/src/bootstrap_model/index/database_index/backfill_state.rs

@@ -15,15 +15,16 @@ pub struct DatabaseIndexBackfillState {
     // before the index was committed because we don't know the commit timestamp.
     // We need to run retention from this timestamp, because live writes write to
     // the index the moment the index committed.
-    pub index_created_lower_bound: Option<Timestamp>,
+    pub index_created_lower_bound: Timestamp,
     // We have done the backfill and the only step left is catch up retention.
     pub retention_started: bool,
 }
 
 #[derive(Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct SerializedDatabaseIndexBackfillState {
-    // TODO: Backfill and remove optional.
+    // NOTE: Those fields should be populated for all latest indexes. We keep them
+    // as option if we ever need to parse historical documents.
     index_created_lower_bound: Option<i64>,
     retention_started: Option<bool>,
 }
@@ -33,7 +34,7 @@ impl TryFrom<DatabaseIndexBackfillState> for SerializedDatabaseIndexBackfillStat
 
     fn try_from(config: DatabaseIndexBackfillState) -> anyhow::Result<Self> {
         Ok(Self {
-            index_created_lower_bound: config.index_created_lower_bound.map(|ts| ts.into()),
+            index_created_lower_bound: Some(config.index_created_lower_bound.into()),
             retention_started: Some(config.retention_started),
         })
     }
@@ -47,8 +48,8 @@ impl TryFrom<SerializedDatabaseIndexBackfillState> for DatabaseIndexBackfillStat
             index_created_lower_bound: config
                 .index_created_lower_bound
                 .map(|ts| ts.try_into())
-                .transpose()?,
-            // Treat legacy records as retention not started.
+                .transpose()?
+                .unwrap_or(Timestamp::MIN),
             retention_started: config.retention_started.unwrap_or(false),
         })
     }

crates/common/src/bootstrap_model/index/database_index/index_state.rs

@@ -4,6 +4,7 @@ use serde::{
     Deserialize,
     Serialize,
 };
+use sync_types::Timestamp;
 use value::codegen_convex_serialization;
 
 use super::{
@@ -67,10 +68,10 @@ impl TryFrom<SerializedDatabaseIndexState> for DatabaseIndexState {
             },
             SerializedDatabaseIndexState::Backfilled2 => DatabaseIndexState::Backfilled,
             SerializedDatabaseIndexState::Enabled => DatabaseIndexState::Enabled,
-            // TODO(Presley): Backfill and delete Disabled state.
+            // None of the latest index documents should be in this state.
             SerializedDatabaseIndexState::Disabled => {
                 DatabaseIndexState::Backfilling(DatabaseIndexBackfillState {
-                    index_created_lower_bound: None,
+                    index_created_lower_bound: Timestamp::MIN,
                     retention_started: false,
                 })
             },

crates/common/src/bootstrap_model/index/index_metadata.rs

@@ -72,7 +72,7 @@ impl<T: TableIdentifier> IndexMetadata<T> {
             config: IndexConfig::Database {
                 developer_config: DeveloperDatabaseIndexConfig { fields },
                 on_disk_state: DatabaseIndexState::Backfilling(DatabaseIndexBackfillState {
-                    index_created_lower_bound: Some(index_created_lower_bound),
+                    index_created_lower_bound,
                     retention_started: false,
                 }),
             },

crates/database/src/index_worker.rs

@@ -68,7 +68,6 @@ use common::{
         RepeatableTimestamp,
         TabletIndexName,
         Timestamp,
-        WriteTimestamp,
     },
     value::{
         ResolvedDocumentId,
@@ -424,8 +423,8 @@ impl<RT: Runtime> IndexWorker<RT> {
         let mut tx = self.database.begin(Identity::system()).await?;
         let index_table_id = tx.bootstrap_tables().index_id;
 
-        let (index_doc, index_ts) = tx
-            .get_with_ts(ResolvedDocumentId::new(index_table_id, index_id))
+        let index_doc = tx
+            .get(ResolvedDocumentId::new(index_table_id, index_id))
             .await?
             .ok_or_else(|| anyhow::anyhow!("Index {index_id:?} no longer exists"))?;
         let mut index_metadata = TabletIndexMetadata::from_document(index_doc)?;
@@ -447,13 +446,10 @@ impl<RT: Runtime> IndexWorker<RT> {
                 };
 
                 state.retention_started = true;
-
-                // TODO(presley): Remove the fallback to commit_ts.
-                let WriteTimestamp::Committed(committed_ts) = index_ts else {
-                    anyhow::bail!("index {index_id} is pending write");
-                };
-                let index_created = state.index_created_lower_bound.unwrap_or(committed_ts);
-                (index_created, developer_config.fields.clone())
+                (
+                    state.index_created_lower_bound,
+                    developer_config.fields.clone(),
+                )
             },
             _ => anyhow::bail!(
                 "IndexWorker attempted to backfill an index {index_metadata:?} which wasn't a \