indexer-alt: kv_feature_flags pipeline

## Description

Like `kv_protocol_configs`, but for feature flags.

## Test plan

Run the indexer over the first 700K checkpoints -- this will include two
protocol config upgrades, then check the database for feature flags.

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

```
sui_indexer_alt=# SELECT
    *
FROM
    kv_feature_flags
ORDER BY
    protocol_version DESC,
    flag_name;
 protocol_version |                          flag_name                          | flag_value
------------------+-------------------------------------------------------------+------------
                6 | accept_zklogin_in_multisig                                  | f
                6 | advance_epoch_start_time_in_safe_mode                       | t
                6 | advance_to_highest_supported_protocol_version               | f
                6 | allow_receiving_object_id                                   | f
                6 | authority_capabilities_v2                                   | f
                                               [...]
```

crates/sui-indexer-alt/migrations/2024-11-02-014903_kv_feature_flags/down.sql

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

crates/sui-indexer-alt/migrations/2024-11-02-014903_kv_feature_flags/up.sql

@@ -0,0 +1,7 @@
+CREATE TABLE IF NOT EXISTS kv_feature_flags
+(
+    protocol_version            BIGINT        NOT NULL,
+    flag_name                   TEXT          NOT NULL,
+    flag_value                  BOOLEAN       NOT NULL,
+    PRIMARY KEY (protocol_version, flag_name)
+);

crates/sui-indexer-alt/src/handlers/kv_feature_flags.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::StoredFeatureFlag},
+    pipeline::{concurrent::Handler, Processor},
+    schema::kv_feature_flags,
+};
+
+pub struct KvFeatureFlags(pub StoredGenesis);
+
+impl Processor for KvFeatureFlags {
+    const NAME: &'static str = "kv_protocol_configs";
+    type Value = StoredFeatureFlag;
+
+    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
+            .feature_map()
+            .into_iter()
+            .map(|(flag_name, flag_value)| StoredFeatureFlag {
+                protocol_version,
+                flag_name,
+                flag_value,
+            })
+            .collect())
+    }
+}
+
+#[async_trait::async_trait]
+impl Handler for KvFeatureFlags {
+    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_feature_flags::table)
+            .values(values)
+            .on_conflict_do_nothing()
+            .execute(conn)
+            .await?)
+    }
+}

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

@@ -4,6 +4,7 @@
 pub mod ev_emit_mod;
 pub mod ev_struct_inst;
 pub mod kv_checkpoints;
+pub mod kv_feature_flags;
 pub mod kv_objects;
 pub mod kv_protocol_configs;
 pub mod kv_transactions;

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_feature_flags::KvFeatureFlags;
 use sui_indexer_alt::handlers::kv_protocol_configs::KvProtocolConfigs;
 use sui_indexer_alt::{
     args::Args,
@@ -42,11 +43,13 @@ async fn main() -> Result<()> {
             let mut indexer = Indexer::new(args.db_config, indexer, cancel.clone()).await?;
 
             let genesis = bootstrap(&indexer, retry_interval, cancel.clone()).await?;
-            let kv_protocol_configs = KvProtocolConfigs(genesis.clone());
+            let kv_feature_flags = KvFeatureFlags(genesis.clone());
+            let kv_protocol_configs = KvProtocolConfigs(genesis);
 
             indexer.concurrent_pipeline(EvEmitMod).await?;
             indexer.concurrent_pipeline(EvStructInst).await?;
             indexer.concurrent_pipeline(KvCheckpoints).await?;
+            indexer.concurrent_pipeline(kv_feature_flags).await?;
             indexer.concurrent_pipeline(KvObjects).await?;
             indexer.concurrent_pipeline(kv_protocol_configs).await?;
             indexer.concurrent_pipeline(KvTransactions).await?;

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

@@ -4,7 +4,15 @@
 use diesel::prelude::*;
 use sui_field_count::FieldCount;
 
-use crate::schema::kv_protocol_configs;
+use crate::schema::{kv_feature_flags, kv_protocol_configs};
+
+#[derive(Insertable, Debug, Clone, FieldCount)]
+#[diesel(table_name = kv_feature_flags)]
+pub struct StoredFeatureFlag {
+    pub protocol_version: i64,
+    pub flag_name: String,
+    pub flag_value: bool,
+}
 
 #[derive(Insertable, Debug, Clone, FieldCount)]
 #[diesel(table_name = kv_protocol_configs)]

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

@@ -30,6 +30,14 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    kv_feature_flags (protocol_version, flag_name) {
+        protocol_version -> Int8,
+        flag_name -> Text,
+        flag_value -> Bool,
+    }
+}
+
 diesel::table! {
     kv_genesis (genesis_digest) {
         genesis_digest -> Bytea,
@@ -205,6 +213,7 @@ diesel::allow_tables_to_appear_in_same_query!(
     ev_emit_mod,
     ev_struct_inst,
     kv_checkpoints,
+    kv_feature_flags,
     kv_genesis,
     kv_objects,
     kv_protocol_configs,