Additional schema changes per pull/25143 (#25438)

nickvikeras · web-flow · commit cd7adf3ac9f1 · 2026-02-13T14:28:01.000-06:00
## Description

Addressing post-merge feedback from pull/25143.

- Additional epoch end fields for graphql
- Optimizing object iteration
- Fixing mislocated pipeline name constants/comments

(Object types table has already been removed, we agreed to just load
types from the object data table directly)

## Test plan

How did you test the new or updated feature?

---

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol:
- [ ] Nodes (Validators and Full nodes):
- [ ] gRPC:
- [ ] JSON-RPC:
- [ ] GraphQL:
- [ ] CLI:
- [ ] Rust SDK:
- [ ] Indexing Framework:
diff --git a/crates/sui-kvstore/src/handlers/epochs_end.rs b/crates/sui-kvstore/src/handlers/epochs_end.rs
@@ -4,15 +4,18 @@
 use std::sync::Arc;
 
 use anyhow::Context as _;
+use anyhow::bail;
 use sui_indexer_alt_framework::pipeline::Processor;
+use sui_types::event::SystemEpochInfoEvent;
 use sui_types::full_checkpoint_content::Checkpoint;
+use sui_types::transaction::TransactionDataAPI;
+use sui_types::transaction::TransactionKind;
 
 use crate::bigtable::proto::bigtable::v2::mutate_rows_request::Entry;
 use crate::handlers::BigTableProcessor;
 use crate::tables;
 
 /// Pipeline that writes epoch end data to BigTable.
-/// This is written when a new epoch starts (for the previous epoch).
 pub struct EpochEndPipeline;
 
 #[async_trait::async_trait]
@@ -21,26 +24,102 @@ impl Processor for EpochEndPipeline {
     type Value = Entry;
 
     async fn process(&self, checkpoint: &Arc<Checkpoint>) -> anyhow::Result<Vec<Self::Value>> {
-        let Some(epoch_info) = checkpoint.epoch_info()? else {
+        let summary = &checkpoint.summary;
+
+        let Some(end_of_epoch) = summary.end_of_epoch_data.as_ref() else {
             return Ok(vec![]);
         };
 
-        if epoch_info.epoch == 0 {
-            return Ok(vec![]);
-        }
+        let Some(transaction) = checkpoint.transactions.iter().find(|tx| {
+            matches!(
+                tx.transaction.kind(),
+                TransactionKind::ChangeEpoch(_) | TransactionKind::EndOfEpochTransaction(_)
+            )
+        }) else {
+            bail!(
+                "Failed to get end of epoch transaction in checkpoint {} with EndOfEpochData",
+                summary.sequence_number,
+            );
+        };
 
-        let epoch_id = epoch_info.epoch - 1;
-        let start_checkpoint = epoch_info
-            .start_checkpoint
-            .context("missing start_checkpoint for epoch end")?;
-        let end_timestamp_ms = epoch_info
-            .start_timestamp_ms
-            .context("missing start_timestamp_ms for epoch end")?;
-        let end_checkpoint = start_checkpoint - 1;
+        let epoch_id = summary.epoch;
+        let end_timestamp_ms = summary.timestamp_ms;
+        let end_checkpoint = summary.sequence_number;
+        let cp_hi = summary.sequence_number + 1;
+        let tx_hi = summary.network_total_transactions;
+
+        let epoch_commitments = bcs::to_bytes(&end_of_epoch.epoch_commitments)
+            .context("Failed to serialize EpochCommitment-s")?;
+
+        let (
+            safe_mode,
+            total_stake,
+            storage_fund_balance,
+            storage_fund_reinvestment,
+            storage_charge,
+            storage_rebate,
+            stake_subsidy_amount,
+            total_gas_fees,
+            total_stake_rewards_distributed,
+            leftover_storage_fund_inflow,
+        ) = if let Some(SystemEpochInfoEvent {
+            total_stake,
+            storage_fund_reinvestment,
+            storage_charge,
+            storage_rebate,
+            storage_fund_balance,
+            stake_subsidy_amount,
+            total_gas_fees,
+            total_stake_rewards_distributed,
+            leftover_storage_fund_inflow,
+            ..
+        }) = transaction
+            .events
+            .iter()
+            .flat_map(|events| &events.data)
+            .find_map(|event| {
+                event
+                    .is_system_epoch_info_event()
+                    .then(|| bcs::from_bytes(&event.contents))
+            })
+            .transpose()
+            .context("Failed to deserialize SystemEpochInfoEvent")?
+        {
+            (
+                false,
+                Some(total_stake),
+                Some(storage_fund_balance),
+                Some(storage_fund_reinvestment),
+                Some(storage_charge),
+                Some(storage_rebate),
+                Some(stake_subsidy_amount),
+                Some(total_gas_fees),
+                Some(total_stake_rewards_distributed),
+                Some(leftover_storage_fund_inflow),
+            )
+        } else {
+            (true, None, None, None, None, None, None, None, None, None)
+        };
 
         let entry = tables::make_entry(
             tables::epochs::encode_key(epoch_id),
-            tables::epochs::encode_end(end_timestamp_ms, end_checkpoint),
+            tables::epochs::encode_end(
+                end_timestamp_ms,
+                end_checkpoint,
+                cp_hi,
+                tx_hi,
+                safe_mode,
+                total_stake,
+                storage_fund_balance,
+                storage_fund_reinvestment,
+                storage_charge,
+                storage_rebate,
+                stake_subsidy_amount,
+                total_gas_fees,
+                total_stake_rewards_distributed,
+                leftover_storage_fund_inflow,
+                &epoch_commitments,
+            ),
             Some(end_timestamp_ms),
         );
 
diff --git a/crates/sui-kvstore/src/handlers/objects.rs b/crates/sui-kvstore/src/handlers/objects.rs
@@ -22,16 +22,18 @@ impl Processor for ObjectsPipeline {
 
     async fn process(&self, checkpoint: &Arc<Checkpoint>) -> anyhow::Result<Vec<Self::Value>> {
         let timestamp_ms = checkpoint.summary.timestamp_ms;
-        let mut entries = Vec::with_capacity(checkpoint.object_set.len());
-
-        for object in checkpoint.object_set.iter() {
-            let object_key = ObjectKey(object.id(), object.version());
-            let entry = tables::make_entry(
-                tables::objects::encode_key(&object_key),
-                tables::objects::encode(object)?,
-                Some(timestamp_ms),
-            );
-            entries.push(entry);
+        let mut entries = vec![];
+
+        for txn in &checkpoint.transactions {
+            for object in txn.output_objects(&checkpoint.object_set) {
+                let object_key = ObjectKey(object.id(), object.version());
+                let entry = tables::make_entry(
+                    tables::objects::encode_key(&object_key),
+                    tables::objects::encode(object)?,
+                    Some(timestamp_ms),
+                );
+                entries.push(entry);
+            }
         }
 
         Ok(entries)
diff --git a/crates/sui-kvstore/src/lib.rs b/crates/sui-kvstore/src/lib.rs
@@ -1,6 +1,39 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
+mod bigtable;
+pub mod config;
+mod handlers;
+pub mod tables;
+
+use std::sync::OnceLock;
+
+use anyhow::Result;
+use async_trait::async_trait;
+use prometheus::Registry;
+use serde::Deserialize;
+use serde::Serialize;
+use sui_indexer_alt_framework::Indexer;
+use sui_indexer_alt_framework::IndexerArgs;
+use sui_indexer_alt_framework::ingestion::ClientArgs;
+use sui_indexer_alt_framework::pipeline::CommitterConfig;
+use sui_indexer_alt_framework::pipeline::concurrent::ConcurrentConfig;
+use sui_types::balance_change::BalanceChange;
+use sui_types::base_types::ObjectID;
+use sui_types::committee::EpochId;
+use sui_types::crypto::AuthorityStrongQuorumSignInfo;
+use sui_types::digests::CheckpointDigest;
+use sui_types::digests::TransactionDigest;
+use sui_types::effects::TransactionEffects;
+use sui_types::effects::TransactionEvents;
+use sui_types::event::Event;
+use sui_types::messages_checkpoint::CheckpointContents;
+use sui_types::messages_checkpoint::CheckpointSequenceNumber;
+use sui_types::messages_checkpoint::CheckpointSummary;
+use sui_types::object::Object;
+use sui_types::storage::ObjectKey;
+use sui_types::transaction::Transaction;
+
 pub use crate::bigtable::client::BigTableClient;
 pub use crate::bigtable::store::BigTableConnection;
 pub use crate::bigtable::store::BigTableStore;
@@ -16,6 +49,11 @@ pub use crate::handlers::ObjectsPipeline;
 pub use crate::handlers::PrevEpochUpdate;
 pub use crate::handlers::TransactionsPipeline;
 pub use crate::handlers::set_max_mutations;
+pub use config::CommitterLayer;
+pub use config::ConcurrentLayer;
+pub use config::IndexerConfig;
+pub use config::IngestionConfig;
+pub use config::PipelineLayer;
 
 pub const CHECKPOINTS_PIPELINE: &str =
     <BigTableHandler<CheckpointsPipeline> as sui_indexer_alt_framework::pipeline::Processor>::NAME;
@@ -32,10 +70,8 @@ pub const EPOCH_END_PIPELINE: &str =
 pub const EPOCH_LEGACY_PIPELINE: &str =
     <EpochLegacyPipeline as sui_indexer_alt_framework::pipeline::Processor>::NAME;
 
-/// All pipeline names registered by the indexer. Single source of truth used for:
-/// - Pipeline registration in `BigTableIndexer::new()`
-/// - Per-pipeline watermark queries in `get_watermark()`
-/// - Legacy watermark tracker expected count
+/// All pipeline names registered by the indexer. Used by `LegacyWatermarkTracker`
+/// to know when all pipelines have reported.
 pub const ALL_PIPELINE_NAMES: [&str; 7] = [
     CHECKPOINTS_PIPELINE,
     CHECKPOINTS_BY_DIGEST_PIPELINE,
@@ -46,44 +82,15 @@ pub const ALL_PIPELINE_NAMES: [&str; 7] = [
     EPOCH_LEGACY_PIPELINE,
 ];
 
-use std::sync::OnceLock;
-
-use anyhow::Result;
-use async_trait::async_trait;
-use prometheus::Registry;
-use serde::Deserialize;
-use serde::Serialize;
-use sui_indexer_alt_framework::Indexer;
-use sui_indexer_alt_framework::IndexerArgs;
-use sui_indexer_alt_framework::ingestion::ClientArgs;
-use sui_indexer_alt_framework::pipeline::CommitterConfig;
-use sui_indexer_alt_framework::pipeline::concurrent::ConcurrentConfig;
-use sui_types::balance_change::BalanceChange;
-use sui_types::base_types::ObjectID;
-use sui_types::committee::EpochId;
-use sui_types::crypto::AuthorityStrongQuorumSignInfo;
-use sui_types::digests::CheckpointDigest;
-use sui_types::digests::TransactionDigest;
-use sui_types::effects::TransactionEffects;
-use sui_types::effects::TransactionEvents;
-use sui_types::event::Event;
-use sui_types::messages_checkpoint::CheckpointContents;
-use sui_types::messages_checkpoint::CheckpointSequenceNumber;
-use sui_types::messages_checkpoint::CheckpointSummary;
-use sui_types::object::Object;
-use sui_types::storage::ObjectKey;
-use sui_types::transaction::Transaction;
-
-mod bigtable;
-pub mod config;
-mod handlers;
-pub mod tables;
-
-pub use config::CommitterLayer;
-pub use config::ConcurrentLayer;
-pub use config::IndexerConfig;
-pub use config::IngestionConfig;
-pub use config::PipelineLayer;
+/// Non-legacy pipeline names used for the default `get_watermark` implementation.
+const WATERMARK_PIPELINES: [&str; 6] = [
+    CHECKPOINTS_PIPELINE,
+    CHECKPOINTS_BY_DIGEST_PIPELINE,
+    TRANSACTIONS_PIPELINE,
+    OBJECTS_PIPELINE,
+    EPOCH_START_PIPELINE,
+    EPOCH_END_PIPELINE,
+];
 
 static WRITE_LEGACY_DATA: OnceLock<bool> = OnceLock::new();
 
@@ -140,6 +147,19 @@ pub struct EpochData {
     pub system_state: Option<sui_types::sui_system_state::SuiSystemState>,
     pub end_timestamp_ms: Option<u64>,
     pub end_checkpoint: Option<u64>,
+    pub cp_hi: Option<u64>,
+    pub tx_hi: Option<u64>,
+    pub safe_mode: Option<bool>,
+    pub total_stake: Option<u64>,
+    pub storage_fund_balance: Option<u64>,
+    pub storage_fund_reinvestment: Option<u64>,
+    pub storage_charge: Option<u64>,
+    pub storage_rebate: Option<u64>,
+    pub stake_subsidy_amount: Option<u64>,
+    pub total_gas_fees: Option<u64>,
+    pub total_stake_rewards_distributed: Option<u64>,
+    pub leftover_storage_fund_inflow: Option<u64>,
+    pub epoch_commitments: Option<Vec<u8>>,
 }
 
 /// Serializable watermark for per-pipeline tracking in BigTable.
@@ -152,16 +172,6 @@ pub struct Watermark {
     pub timestamp_ms_hi_inclusive: u64,
 }
 
-/// Non-legacy pipeline names used for the default `get_watermark` implementation.
-const WATERMARK_PIPELINES: [&str; 6] = [
-    CHECKPOINTS_PIPELINE,
-    CHECKPOINTS_BY_DIGEST_PIPELINE,
-    TRANSACTIONS_PIPELINE,
-    OBJECTS_PIPELINE,
-    EPOCH_START_PIPELINE,
-    EPOCH_END_PIPELINE,
-];
-
 #[async_trait]
 pub trait KeyValueStoreReader {
     async fn get_objects(&mut self, objects: &[ObjectKey]) -> Result<Vec<Object>>;
diff --git a/crates/sui-kvstore/src/tables/epochs.rs b/crates/sui-kvstore/src/tables/epochs.rs
