feat: expose retained block range

keeper/src/accessor.rs

@@ -1,6 +1,11 @@
 //! Namespaced keeper access APIs.
 
-use std::{cell::LazyCell, ops::Deref, path::PathBuf, sync::Arc};
+use std::{
+    cell::LazyCell,
+    ops::{Deref, RangeInclusive},
+    path::PathBuf,
+    sync::Arc,
+};
 
 use accountsdb::{AccountEntry, AccountsDB, AccountsDBError};
 use ledger::{
@@ -310,6 +315,11 @@ impl<'a> BlocksAccessor<'a> {
         Ok(request(self.keeper, params, ReadRequest::Block).await??)
     }
 
+    /// Returns published retained bounds, assuming consecutive block slots.
+    pub fn retained_range(&self) -> Option<RangeInclusive<Slot>> {
+        self.keeper.ledger.retained_range()
+    }
+
     /// Returns the latest block boundary known to keeper.
     pub fn latest(&self) -> Block {
         **self.keeper.caches.blocks.latest.load()

ledger/src/lib.rs

@@ -3,6 +3,7 @@
 use std::{
     collections::BTreeMap,
     fs::{self, File},
+    ops::RangeInclusive,
     path::{Path, PathBuf},
     sync::{Arc, atomic::Ordering::*},
     thread::{self, JoinHandle},
@@ -133,6 +134,16 @@ impl Ledger {
         range.rev().filter_map(|id| self.superblocks.read().get(&id).cloned())
     }
 
+    /// Returns the published retained slot bounds for consecutive-slot producers.
+    /// An empty ledger has no range; publication does not fence the asynchronous index.
+    pub fn retained_range(&self) -> Option<RangeInclusive<Slot>> {
+        let superblocks = self.superblocks.read();
+        let (_, first) = superblocks.first_key_value()?;
+        let end = self.meta.range.end.load(Acquire);
+        let start = first.meta.range.start.load(Acquire);
+        (start != 0 && start <= end).then_some(start..=end)
+    }
+
     /// Returns the highest slot recorded across retained superblocks, or
     /// `None` when none are retained. Taken over all superblocks rather than
     /// just the head: right after a seal the active head has no blocks yet and

ledger/src/tests/integration.rs

@@ -489,12 +489,19 @@ async fn test_block_detail_levels_partition_transactions() {
 async fn test_superblock_rotation_and_retention() {
     // size_limit 0 makes every block boundary trigger a retention pass.
     let (dir, ledger) = ledger(0);
+    assert_eq!(ledger.retained_range(), None);
     let old = block_of(&ledger, 1, 1);
+    assert_eq!(ledger.retained_range(), Some(1..=1));
     // Seal superblock 1 and rotate to superblock 2.
     let events = vec![seal(&ledger, 1)];
     append(&ledger, events);
     assert_eq!(ledger.meta.head(), 2);
 
+    assert_eq!(
+        ledger.retained_range(),
+        Some(1..=1),
+        "empty head preserves the retained tip"
+    );
     // The sealed superblock is still readable through the newer head.
     assert!(read_transaction(&ledger, old[0]).await.is_some());
 
@@ -505,6 +512,10 @@ async fn test_superblock_rotation_and_retention() {
         ledger.superblocks.read().get(&1).is_none(),
         "oldest sealed segment purged"
     );
+    assert_eq!(
+        ledger.retained_range(),
+        Some(SIZE_CHECK_FREQUENCY..=SIZE_CHECK_FREQUENCY)
+    );
     // Its transactions are gone; the head's remain.
     assert!(read_transaction(&ledger, old[0]).await.is_none());
     assert!(read_transaction(&ledger, new[0]).await.is_some());
@@ -530,6 +541,7 @@ async fn test_block_read_across_superblocks() {
     append(&ledger, events);
     block_of(&ledger, 2, 1);
 
+    assert_eq!(ledger.retained_range(), Some(1..=2));
     // Slot 1 lives in the sealed superblock; slot 2 in the head. Both resolve.
     assert_eq!(block_signatures(&ledger, 1).await, first);
     assert!(read_block(&ledger, 2, BlockDetails::None).await.is_some());