Add ledger-tool bigtable upload loop (backport #26030)

ledger-tool/src/bigtable.rs

@@ -25,6 +25,7 @@ use {
         UiTransactionEncoding,
     },
     std::{
+        cmp::min,
         collections::HashSet,
         path::Path,
         process::exit,
@@ -36,7 +37,7 @@ use {
 
 async fn upload(
     blockstore: Blockstore,
-    starting_slot: Slot,
+    mut starting_slot: Slot,
     ending_slot: Option<Slot>,
     force_reupload: bool,
     config: solana_storage_bigtable::LedgerStorageConfig,
@@ -49,19 +50,29 @@ async fn upload(
         force_reupload,
         ..ConfirmedBlockUploadConfig::default()
     };
+    let blockstore = Arc::new(blockstore);
+
+    let ending_slot = ending_slot.unwrap_or_else(|| blockstore.last_root());
 
-    solana_ledger::bigtable_upload::upload_confirmed_blocks(
-        Arc::new(blockstore),
-        bigtable,
-        starting_slot,
-        ending_slot,
-        config,
-        Arc::new(AtomicBool::new(false)),
-    )
-    .await
-    .map(|last_slot_uploaded| {
+    while starting_slot <= ending_slot {
+        let current_ending_slot = min(
+            ending_slot,
+            starting_slot.saturating_add(config.max_num_slots_to_check as u64 * 2),
+        );
+        let last_slot_uploaded = solana_ledger::bigtable_upload::upload_confirmed_blocks(
+            blockstore.clone(),
+            bigtable.clone(),
+            starting_slot,
+            current_ending_slot,
+            config.clone(),
+            Arc::new(AtomicBool::new(false)),
+        )
+        .await?;
         info!("last slot uploaded: {}", last_slot_uploaded);
-    })
+        starting_slot = last_slot_uploaded.saturating_add(1);
+    }
+    info!("No more blocks to upload.");
+    Ok(())
 }
 
 async fn delete_slots(

ledger/src/bigtable_upload.rs

@@ -45,7 +45,7 @@ pub async fn upload_confirmed_blocks(
     blockstore: Arc<Blockstore>,
     bigtable: solana_storage_bigtable::LedgerStorage,
     starting_slot: Slot,
-    ending_slot: Option<Slot>,
+    ending_slot: Slot,
     config: ConfirmedBlockUploadConfig,
     exit: Arc<AtomicBool>,
 ) -> Result<Slot, Box<dyn std::error::Error>> {
@@ -60,14 +60,7 @@ pub async fn upload_confirmed_blocks(
                 starting_slot, err
             )
         })?
-        .map_while(|slot| {
-            if let Some(ending_slot) = &ending_slot {
-                if slot > *ending_slot {
-                    return None;
-                }
-            }
-            Some(slot)
-        })
+        .map_while(|slot| (slot <= ending_slot).then(|| slot))
         .collect();
 
     if blockstore_slots.is_empty() {

ledger/src/bigtable_upload_service.rs

@@ -102,7 +102,7 @@ impl BigTableUploadService {
                 blockstore.clone(),
                 bigtable_ledger_storage.clone(),
                 start_slot,
-                Some(end_slot),
+                end_slot,
                 config.clone(),
                 exit.clone(),
             ));