indexer-alt: kv_protocol_configs pipeline

## Description

Add a pipeline for indexing protocol configs.

## Test plan

Index the first ~1M checkpoints, which should pick up the first
protocol version upgrade (from 4 to 5 for epoch 10) and then query
`kv_protocol_configs` results after that:

```
sui$ cargo run -p sui-indexer-alt --                                           \
  --database-url postgres://postgres:postgrespw@localhost:5432/sui_indexer_alt \
  --remote-store-url https://checkpoints.mainnet.sui.io                        \
  --last-checkpoint 500000
```

```
sui_indexer_alt=# SELECT * FROM kv_protocol_configs;
```

crates/sui-indexer-alt/migrations/2024-11-01-153859_kv_protocol_configs/down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS kv_protocol_configs;

crates/sui-indexer-alt/migrations/2024-11-01-153859_kv_protocol_configs/up.sql

@@ -0,0 +1,7 @@
+CREATE TABLE IF NOT EXISTS kv_protocol_configs
+(
+    protocol_version            BIGINT        NOT NULL,
+    config_name                 TEXT          NOT NULL,
+    config_value                TEXT,
+    PRIMARY KEY (protocol_version, config_name)
+);

crates/sui-indexer-alt/src/handlers/kv_protocol_configs.rs

@@ -0,0 +1,68 @@
+// Copyright (c) Mysten Labs, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+use std::sync::Arc;
+
+use anyhow::{Context, Result};
+use diesel_async::RunQueryDsl;
+use sui_protocol_config::ProtocolConfig;
+use sui_types::full_checkpoint_content::CheckpointData;
+
+use crate::{
+    db,
+    models::{checkpoints::StoredGenesis, epochs::StoredProtocolConfig},
+    pipeline::{concurrent::Handler, Processor},
+    schema::kv_protocol_configs,
+};
+
+pub struct KvProtocolConfigs(pub StoredGenesis);
+
+impl Processor for KvProtocolConfigs {
+    const NAME: &'static str = "kv_protocol_configs";
+    type Value = StoredProtocolConfig;
+
+    fn process(&self, checkpoint: &Arc<CheckpointData>) -> Result<Vec<Self::Value>> {
+        let CheckpointData {
+            checkpoint_summary, ..
+        } = checkpoint.as_ref();
+
+        let protocol_version = if checkpoint_summary.sequence_number == 0 {
+            self.0.initial_protocol_version()
+        } else if let Some(end_of_epoch) = checkpoint_summary.end_of_epoch_data.as_ref() {
+            end_of_epoch.next_epoch_protocol_version
+        } else {
+            return Ok(vec![]);
+        };
+
+        let protocol_config = ProtocolConfig::get_for_version(
+            protocol_version,
+            self.0.chain().context("Failed to identify chain")?,
+        );
+
+        let protocol_version = protocol_version.as_u64() as i64;
+        Ok(protocol_config
+            .attr_map()
+            .into_iter()
+            .map(|(config_name, value)| StoredProtocolConfig {
+                protocol_version,
+                config_name,
+                config_value: value.map(|v| v.to_string()),
+            })
+            .collect())
+    }
+}
+
+#[async_trait::async_trait]
+impl Handler for KvProtocolConfigs {
+    const MIN_EAGER_ROWS: usize = 1;
+    const MAX_CHUNK_ROWS: usize = i16::MAX as usize / 3;
+    const MAX_PENDING_ROWS: usize = 10000;
+
+    async fn commit(values: &[Self::Value], conn: &mut db::Connection<'_>) -> Result<usize> {
+        Ok(diesel::insert_into(kv_protocol_configs::table)
+            .values(values)
+            .on_conflict_do_nothing()
+            .execute(conn)
+            .await?)
+    }
+}

crates/sui-indexer-alt/src/handlers/mod.rs

@@ -5,6 +5,7 @@ pub mod ev_emit_mod;
 pub mod ev_struct_inst;
 pub mod kv_checkpoints;
 pub mod kv_objects;
+pub mod kv_protocol_configs;
 pub mod kv_transactions;
 pub mod obj_versions;
 pub mod sum_coin_balances;

crates/sui-indexer-alt/src/main.rs

@@ -6,6 +6,7 @@ use clap::Parser;
 use sui_indexer_alt::args::Command;
 use sui_indexer_alt::bootstrap::bootstrap;
 use sui_indexer_alt::db::reset_database;
+use sui_indexer_alt::handlers::kv_protocol_configs::KvProtocolConfigs;
 use sui_indexer_alt::{
     args::Args,
     handlers::{
@@ -40,12 +41,14 @@ async fn main() -> Result<()> {
             let retry_interval = indexer.ingestion_config.retry_interval;
             let mut indexer = Indexer::new(args.db_config, indexer, cancel.clone()).await?;
 
-            bootstrap(&indexer, retry_interval, cancel.clone()).await?;
+            let genesis = bootstrap(&indexer, retry_interval, cancel.clone()).await?;
+            let kv_protocol_configs = KvProtocolConfigs(genesis.clone());
 
             indexer.concurrent_pipeline(EvEmitMod).await?;
             indexer.concurrent_pipeline(EvStructInst).await?;
             indexer.concurrent_pipeline(KvCheckpoints).await?;
             indexer.concurrent_pipeline(KvObjects).await?;
+            indexer.concurrent_pipeline(kv_protocol_configs).await?;
             indexer.concurrent_pipeline(KvTransactions).await?;
             indexer.concurrent_pipeline(ObjVersions).await?;
             indexer.concurrent_pipeline(TxAffectedAddress).await?;

crates/sui-indexer-alt/src/models/epochs.rs

@@ -0,0 +1,15 @@
+// Copyright (c) Mysten Labs, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+use diesel::prelude::*;
+use sui_field_count::FieldCount;
+
+use crate::schema::kv_protocol_configs;
+
+#[derive(Insertable, Debug, Clone, FieldCount)]
+#[diesel(table_name = kv_protocol_configs)]
+pub struct StoredProtocolConfig {
+    pub protocol_version: i64,
+    pub config_name: String,
+    pub config_value: Option<String>,
+}

crates/sui-indexer-alt/src/models/mod.rs

@@ -3,6 +3,7 @@
 
 pub mod checkpoints;
 pub mod displays;
+pub mod epochs;
 pub mod events;
 pub mod objects;
 pub mod packages;

crates/sui-indexer-alt/src/schema.rs

@@ -45,6 +45,14 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    kv_protocol_configs (protocol_version, config_name) {
+        protocol_version -> Int8,
+        config_name -> Text,
+        config_value -> Nullable<Text>,
+    }
+}
+
 diesel::table! {
     kv_transactions (tx_digest) {
         tx_digest -> Bytea,
@@ -199,6 +207,7 @@ diesel::allow_tables_to_appear_in_same_query!(
     kv_checkpoints,
     kv_genesis,
     kv_objects,
+    kv_protocol_configs,
     kv_transactions,
     obj_versions,
     sum_coin_balances,