feat(core): live/historic storage split with per-epoch history buckets

crates/iota-config/data/fullnode-template-with-path.yaml

@@ -19,7 +19,9 @@ migration-tx-data-path: "migration.blob"
 
 authority-store-pruning-config:
   num-latest-epoch-dbs-to-retain: 3
-  num-epochs-to-retain: 1
+  # Number of epochs of superseded object versions and pruned checkpoint
+  # data to keep readable; whole epoch buckets are dropped past this window.
+  historic-epochs-to-retain: 2
 
 authority-key-pair:
   path: "authority.key"

crates/iota-config/data/fullnode-template.yaml

@@ -19,4 +19,6 @@ migration-tx-data-path: "/opt/iota/config/migration.blob"
 
 authority-store-pruning-config:
   num-latest-epoch-dbs-to-retain: 3
-  num-epochs-to-retain: 1
+  # Number of epochs of superseded object versions and pruned checkpoint
+  # data to keep readable; whole epoch buckets are dropped past this window.
+  historic-epochs-to-retain: 2

crates/iota-config/src/node.rs

@@ -1030,12 +1030,6 @@ pub struct AuthorityStorePruningConfig {
     /// number of the latest epoch dbs to retain
     #[serde(default = "default_num_latest_epoch_dbs_to_retain")]
     pub num_latest_epoch_dbs_to_retain: usize,
-    /// number of epochs to keep the latest version of objects for.
-    /// Note that a zero value corresponds to an aggressive pruner.
-    /// This mode is experimental and needs to be used with caution.
-    /// Use `u64::MAX` to disable the pruner for the objects.
-    #[serde(default)]
-    pub num_epochs_to_retain: u64,
     /// enables periodic background compaction for old SST files whose last
     /// modified time is older than `periodic_compaction_threshold_days`
     /// days. That ensures that all sst files eventually go through the
@@ -1049,15 +1043,21 @@ pub struct AuthorityStorePruningConfig {
     /// for
     #[serde(skip_serializing_if = "Option::is_none")]
     pub num_epochs_to_retain_for_checkpoints: Option<u64>,
-    /// Enables the compaction filter for pruning the objects table.
-    /// If disabled, a range deletion approach is used instead.
-    /// While it is generally safe to switch between the two modes,
-    /// switching from the compaction filter approach back to range deletion
-    /// may result in some old versions that will never be pruned.
-    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
-    pub enable_compaction_filter: bool,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub num_epochs_to_retain_for_indexes: Option<u64>,
+    /// Number of epochs of historic data to retain: superseded object
+    /// versions and pruned checkpoint-keyed history (transactions, effects,
+    /// events, checkpoint contents and summaries) are relocated into
+    /// per-epoch buckets, bucketed by the epoch in which they were
+    /// superseded, and whole buckets are dropped once they fall out of this
+    /// window. Historic data remains readable through exact-version gRPC
+    /// lookups and checkpoint reads until then.
+    #[serde(default = "default_historic_epochs_to_retain")]
+    pub historic_epochs_to_retain: u64,
+}
+
+fn default_historic_epochs_to_retain() -> u64 {
+    2
 }
 
 fn default_num_latest_epoch_dbs_to_retain() -> usize {
@@ -1072,20 +1072,15 @@ impl Default for AuthorityStorePruningConfig {
     fn default() -> Self {
         Self {
             num_latest_epoch_dbs_to_retain: default_num_latest_epoch_dbs_to_retain(),
-            num_epochs_to_retain: 0,
             periodic_compaction_threshold_days: None,
             num_epochs_to_retain_for_checkpoints: if cfg!(msim) { Some(2) } else { None },
-            enable_compaction_filter: cfg!(test) || cfg!(msim),
             num_epochs_to_retain_for_indexes: None,
+            historic_epochs_to_retain: default_historic_epochs_to_retain(),
         }
     }
 }
 
 impl AuthorityStorePruningConfig {
-    pub fn set_num_epochs_to_retain(&mut self, num_epochs_to_retain: u64) {
-        self.num_epochs_to_retain = num_epochs_to_retain;
-    }
-
     pub fn set_num_epochs_to_retain_for_checkpoints(&mut self, num_epochs_to_retain: Option<u64>) {
         self.num_epochs_to_retain_for_checkpoints = num_epochs_to_retain;
     }

crates/iota-core/src/authority.rs

@@ -157,8 +157,8 @@ use crate::{
         authority_per_epoch_store_pruner::AuthorityPerEpochStorePruner,
         authority_store::{ExecutionLockReadGuard, ObjectLockStatus},
         authority_store_pruner::{AuthorityStorePruner, EPOCH_DURATION_MS_FOR_TESTING},
-        authority_store_tables::AuthorityPrunerTables,
         epoch_start_configuration::{EpochStartConfigTrait, EpochStartConfiguration},
+        historic_store::HistoricStore,
     },
     authority_client::NetworkAuthorityClient,
     checkpoint_progress_tracker::CheckpointProgressTracker,
@@ -233,6 +233,7 @@ pub mod authority_store_pruner;
 pub mod authority_store_tables;
 pub mod authority_store_types;
 pub mod epoch_start_configuration;
+pub mod historic_store;
 pub mod shared_object_congestion_tracker;
 pub mod shared_object_version_manager;
 pub mod suggested_gas_price_calculator;
@@ -853,6 +854,11 @@ pub struct AuthorityState {
     pub indexes: Option<Arc<IndexStore>>,
     pub grpc_indexes_store: Option<Arc<GrpcIndexesStore>>,
 
+    /// Superseded object versions relocated out of the live objects table.
+    /// Read exclusively by the gRPC exact-version object lookup; consensus
+    /// and execution paths must never consult it.
+    pub historic_store: Arc<HistoricStore>,
+
     pub subscription_handler: Arc<SubscriptionHandler>,
     pub checkpoint_store: Arc<CheckpointStore>,
 
@@ -867,7 +873,7 @@ pub struct AuthorityState {
 
     pub metrics: Arc<AuthorityMetrics>,
     /// The store pruner. The checkpoint executor uses it to nudge the pruner
-    /// after each checkpoint and to be leashed if pruning falls behind.
+    /// after each checkpoint.
     pruner: AuthorityStorePruner,
     authority_per_epoch_pruner: AuthorityPerEpochStorePruner,
     checkpoint_progress_tracker: Option<Arc<CheckpointProgressTracker>>,
@@ -1720,11 +1726,39 @@ impl AuthorityState {
         // Allow testing what happens if we crash here.
         fail_point!("crash");
 
-        let transaction_outputs = TransactionOutputs::build_transaction_outputs(
+        let mut transaction_outputs = TransactionOutputs::build_transaction_outputs(
             transaction.clone().into_unsigned(),
             effects.clone(),
             inner_temporary_store,
         );
+        // Mutations of runtime-loaded objects (dynamic fields) are not
+        // transaction inputs, so their pre-images could not be captured from
+        // `input_objects`. Read them back through the cache — the transaction
+        // just read them, so they are memory-hot — to complete the capture;
+        // a version already relocated is absent from the live view and needs
+        // no move.
+        let captured: HashSet<ObjectKey> = transaction_outputs
+            .superseded
+            .iter()
+            .map(|(key, _)| *key)
+            .collect();
+        let missing: Vec<ObjectKey> = effects
+            .modified_at_versions()
+            .into_iter()
+            .map(|(object_id, version)| ObjectKey(object_id, version))
+            .filter(|key| !captured.contains(key))
+            .collect();
+        if !missing.is_empty() {
+            let objects = self
+                .get_object_cache_reader()
+                .try_multi_get_objects_by_key(&missing)?;
+            transaction_outputs.superseded.extend(
+                missing
+                    .into_iter()
+                    .zip(objects)
+                    .filter_map(|(key, object)| object.map(|object| (key, object))),
+            );
+        }
         self.get_cache_writer()
             .try_write_transaction_outputs(epoch_store.epoch(), transaction_outputs.into())?;
 
@@ -3258,7 +3292,7 @@ impl AuthorityState {
         archive_readers: ArchiveReaderBalancer,
         validator_tx_finalizer: Option<Arc<ValidatorTxFinalizer<NetworkAuthorityClient>>>,
         chain_identifier: ChainIdentifier,
-        pruner_db: Option<Arc<AuthorityPrunerTables>>,
+        historic_store: Arc<HistoricStore>,
         checkpoint_progress_tracker: Option<Arc<CheckpointProgressTracker>>,
         policy_config: Option<PolicyConfig>,
         firewall_config: Option<RemoteFirewallConfig>,
@@ -3291,11 +3325,10 @@ impl AuthorityState {
             grpc_indexes_store.clone(),
             indexes.clone(),
             config.authority_store_pruning_config.clone(),
-            epoch_store.committee().authority_exists(&name),
             epoch_store.epoch_start_state().epoch_duration_ms(),
             prometheus_registry,
             archive_readers,
-            pruner_db,
+            historic_store.clone(),
             checkpoint_progress_tracker.clone(),
         );
         let input_loader =
@@ -3325,6 +3358,7 @@ impl AuthorityState {
             execution_cache_trait_pointers,
             indexes,
             grpc_indexes_store,
+            historic_store,
             subscription_handler: Arc::new(SubscriptionHandler::new(prometheus_registry)),
             checkpoint_store,
             committee_store,
@@ -3426,8 +3460,8 @@ impl AuthorityState {
             &self.database_for_testing().perpetual_tables,
             &self.checkpoint_store,
             self.grpc_indexes_store.as_deref(),
-            None,
-            config.authority_store_pruning_config,
+            &self.historic_store,
+            &config.authority_store_pruning_config,
             metrics,
             archive_readers,
             EPOCH_DURATION_MS_FOR_TESTING,
@@ -3846,6 +3880,10 @@ impl AuthorityState {
             }
         }
 
+        // The historic epoch buckets are column families of the perpetual
+        // database, so the perpetual snapshot above already covers them
+        // consistently — no separate history snapshot is needed.
+
         fs::rename(checkpoint_path_tmp, checkpoint_path)
             .map_err(|e| IotaError::FileIO(e.to_string()))?;
         Ok(())
@@ -4065,11 +4103,22 @@ impl AuthorityState {
         object_id: &ObjectId,
         version: SequenceNumber,
     ) -> IotaResult<Option<(Object, Option<MoveStructLayout>)>> {
-        let Some(object) = self
+        // Superseded versions leave the live table when their checkpoint
+        // commits, so exact-version reads fall back to the historic buckets.
+        // This is a read-API path only; consensus and execution never take
+        // it.
+        let object = match self
             .get_object_cache_reader()
             .try_get_object_by_key(object_id, version)?
-        else {
-            return Ok(None);
+        {
+            Some(object) => object,
+            None => match self
+                .historic_store
+                .get_object(&ObjectKey(*object_id, version))?
+            {
+                Some(object) => object,
+                None => return Ok(None),
+            },
         };
 
         let layout = self.get_object_layout(&object)?;
@@ -4277,16 +4326,49 @@ impl AuthorityState {
         &self,
         effects: &TransactionEffects,
     ) -> anyhow::Result<Vec<Object>> {
-        iota_types::storage::get_transaction_input_objects(self.get_object_store(), effects)
-            .map_err(Into::into)
+        let input_object_keys: Vec<ObjectKey> = effects
+            .modified_at_versions()
+            .into_iter()
+            .map(|(object_id, version)| ObjectKey(object_id, version))
+            .collect();
+        self.multi_get_objects_with_historic_fallback(&input_object_keys, effects)
     }
 
     pub fn get_transaction_output_objects(
         &self,
         effects: &TransactionEffects,
     ) -> anyhow::Result<Vec<Object>> {
-        iota_types::storage::get_transaction_output_objects(self.get_object_store(), effects)
-            .map_err(Into::into)
+        let output_object_keys: Vec<ObjectKey> = effects
+            .all_changed_objects()
+            .into_iter()
+            .map(|(object_ref, _owner, _kind)| ObjectKey::from(object_ref))
+            .collect();
+        self.multi_get_objects_with_historic_fallback(&output_object_keys, effects)
+    }
+
+    /// Exact-version reads for response assembly: live table first, historic
+    /// buckets second. A transaction's own commit relocates the versions it
+    /// superseded, so responses assembled after the commit must be able to
+    /// reach them. Read-API only; execution never takes this path.
+    fn multi_get_objects_with_historic_fallback(
+        &self,
+        keys: &[ObjectKey],
+        effects: &TransactionEffects,
+    ) -> anyhow::Result<Vec<Object>> {
+        self.get_object_store()
+            .multi_get_objects_by_key(keys)
+            .into_iter()
+            .zip(keys)
+            .map(|(maybe_object, key)| match maybe_object {
+                Some(object) => Ok(object),
+                None => self.historic_store.get_object(key)?.ok_or_else(|| {
+                    anyhow::anyhow!(
+                        "missing object key {key:?} from tx {}",
+                        effects.transaction_digest()
+                    )
+                }),
+            })
+            .collect()
     }
 
     fn get_indexes(&self) -> IotaResult<Arc<IndexStore>> {
@@ -4350,7 +4432,7 @@ impl AuthorityState {
     }
 
     /// The store pruner; the checkpoint executor uses it to nudge the pruner
-    /// after each checkpoint and to be leashed when pruning falls behind.
+    /// after each checkpoint.
     pub fn pruner(&self) -> &AuthorityStorePruner {
         &self.pruner
     }
@@ -6071,18 +6153,33 @@ impl TransactionKeyValueStoreTrait for AuthorityState {
         object_id: ObjectId,
         version: VersionNumber,
     ) -> IotaResult<Option<Object>> {
-        self.get_object_cache_reader()
-            .try_get_object_by_key(&object_id, version)
+        // Exact-version serving read: superseded versions live in the
+        // historic buckets after their checkpoint committed.
+        match self
+            .get_object_cache_reader()
+            .try_get_object_by_key(&object_id, version)?
+        {
+            Some(object) => Ok(Some(object)),
+            None => self
+                .historic_store
+                .get_object(&ObjectKey(object_id, version)),
+        }
     }
 
     #[instrument(skip_all)]
     async fn multi_get_objects(
         &self,
         object_keys: &[ObjectKey],
     ) -> IotaResult<Vec<Option<Object>>> {
-        Ok(self
+        let mut objects = self
             .get_object_cache_reader()
-            .multi_get_objects_by_key(object_keys))
+            .multi_get_objects_by_key(object_keys);
+        for (maybe_object, key) in objects.iter_mut().zip(object_keys) {
+            if maybe_object.is_none() {
+                *maybe_object = self.historic_store.get_object(key)?;
+            }
+        }
+        Ok(objects)
     }
 
     async fn multi_get_transactions_perpetual_checkpoints(