Closing settlements (#54)

* [settlement-pipelines] close settlements

* [settlement-pipelines] refactoring for cleaner code

* [settlement-pipelines] review changes

.buildkite/claim-settlements.yml

@@ -57,7 +57,7 @@ steps:
 
   - label: ":campfire: Claim settlements"
     env:
-      RUST_LOG: solana_transaction_builder_executor=debug,solana_transaction_builder=debug,builder_executor=debug,solana_transaction_executor=debug,settlement_pipelines=debug,claim_settlement=debug
+      RUST_LOG: info,solana_transaction_builder_executor=debug,solana_transaction_builder=debug,builder_executor=debug,solana_transaction_executor=debug,settlement_pipelines=debug,claim_settlement=debug
       # RUST_BACKTRACE: full
     commands:
     - |
@@ -69,9 +69,10 @@ steps:
     - 'chmod +x target/release/claim-settlement'
     - 'echo "UNKNOWN ERROR" > ./claiming-report.txt'
     - |
+      set -o pipefail
       ./target/release/claim-settlement \
         --rpc-url $$RPC_URL \
-        --merkle-trees-dir "./merkle-trees/" \
+        --settlement-json-files ./merkle-trees/* \
         --operator-authority "$$VALIDATOR_BONDS_OPERATOR_AUTHORITY" \
         --fee-payer "$$VALIDATOR_BONDS_RANDOM_TX_FEE_PAYER" \
         --rent-payer "$$VALIDATOR_BONDS_RANDOM_RENT_PAYER" \

.buildkite/close-settlements.yml

@@ -0,0 +1,107 @@
+agents:
+  queue: "snapshots"
+
+steps:
+  - command: echo "--> Start of concurrency gate"
+    concurrency_group: 'validator-bonds/close-settlements'
+    concurrency: 1
+
+  - wait: ~
+
+  - label: ":hammer_and_wrench: :rust: Build"
+    commands:
+    - '. "$HOME/.cargo/env"'
+    - 'cargo build --release --bin close-settlement'
+    - 'cargo build --release --bin list-settlement'
+    artifact_paths:
+      - target/release/close-settlement
+      - target/release/list-settlement
+
+  - label: " Loading past settlements data"
+    env:
+      gs_bucket: gs://marinade-validator-bonds-mainnet
+      config_epochs_non_closable: 3 # configured onchain in config
+      past_epochs_to_load: 5
+    commands:
+    - |
+      set -x
+      current_epoch=$(curl --silent "$$RPC_URL" -X POST -H "Content-Type: application/json" -d '{"jsonrpc":"2.0","id":1, "method":"getEpochInfo"}' | jq '.result.epoch')
+      epochs_end_index=$((current_epoch - config_epochs_non_closable))
+      epochs_start_index=$((epochs_end_index - past_epochs_to_load))
+    - 'mkdir ./merkle-trees/'
+    - |
+      if [[ $$epochs_start_index -lt 1 ]]; then
+        echo "No found any epoch to start to load settlement JSON files"
+        exit 1
+      fi
+    - |
+      set -x
+      for epoch in $(seq $$epochs_start_index $$epochs_end_index); do
+        gcloud storage cp "$$gs_bucket/$$epoch/settlement-merkle-trees.json" "./merkle-trees/$${epoch}_settlement-merkle-trees.json"
+      done
+    artifact_paths:
+      - "./merkle-trees/*"
+
+  - wait: ~
+
+  - label: ":campfire: List past settlements"
+    env:
+      RUST_LOG: info,solana_transaction_builder_executor=debug,solana_transaction_builder=debug,solana_transaction_executor=debug,list_settlement=debug
+    commands:
+    - '. "$HOME/.cargo/env"'
+    - 'buildkite-agent artifact download --include-retried-jobs "merkle-trees/*" .'
+    - 'buildkite-agent artifact download --include-retried-jobs target/release/list-settlement .'
+    - 'chmod +x target/release/list-settlement'
+    - './target/release/list-settlement -u $$RPC_URL -m ./merkle-trees/* --out ./past-settlements.json'
+    artifact_paths:
+      - "./past-settlements.json"
+
+  - wait: ~
+
+  - label: ":campfire: Close settlements"
+    key: 'close-settlement'
+    env:
+      RUST_LOG: info,solana_transaction_builder_executor=debug,solana_transaction_builder=debug,solana_transaction_executor=debug,close_settlement=debug
+      # RUST_BACKTRACE: full
+    commands:
+    - '. "$HOME/.cargo/env"'
+    - 'buildkite-agent artifact download --include-retried-jobs target/release/close-settlement .'
+    - 'buildkite-agent artifact download --include-retried-jobs past-settlements.json .'
+    - 'chmod +x target/release/close-settlement'
+    - |
+      set -o pipefail
+      ./target/release/close-settlement \
+        --rpc-url $$RPC_URL \
+        --operator-authority "$$VALIDATOR_BONDS_OPERATOR_AUTHORITY" \
+        --fee-payer "$$VALIDATOR_BONDS_FUNDING_WALLET" \
+        --marinade-wallet "$$VALIDATOR_BONDS_FUNDING_WALLET" \
+        --past-settlements ./past-settlements.json | tee close-settlement-report.txt
+    artifact_paths:
+    - "./close-settlement-report.txt"
+
+  # discord reporting only in case of failure => manual intervention is needed
+  - label: ":mega: Notification settlements closing"
+    commands:
+    - 'build_result=$(buildkite-agent step get "outcome" --step "close-settlement")'
+    - |
+        if [[ "$$build_result" =~ "failed" ]]; then
+          buildkite-agent artifact download --include-retried-jobs close-settlement-report.txt . || echo 'UNKNOWN ERROR' > './close-settlement-report.txt'
+          curl "$$DISCORD_WEBHOOK_VALIDATOR_BONDS" \
+            -F 'payload_json={
+                "content": "Closed Settlements FAILURE, epoch: '"$$claimable_epochs"'",
+                "url": "'"$$BUILDKITE_BUILD_URL"'",
+                "color": "3158271"
+            }' \
+            -F "file1=@./close-settlement-report.txt"
+        else
+          echo "No error, no need to notify about closing settlements, all good"
+        fi
+
+    depends_on: "close-settlement"
+    allow_dependency_failure: true
+
+  - wait: ~
+
+  - command: echo "End of concurrency gate <--"
+    concurrency_group: 'validator-bonds/close-settlements'
+    concurrency: 1

.buildkite/init-settlements.yml

@@ -75,7 +75,7 @@ steps:
 
   - label: ":campfire: Create settlements"
     env:
-      RUST_LOG: solana_transaction_builder_executor=debug,solana_transaction_builder=debug,builder_executor=debug,solana_transaction_executor=debug,settlement_pipelines=debug,init_settlement=debug
+      RUST_LOG: info,solana_transaction_builder_executor=debug,solana_transaction_builder=debug,builder_executor=debug,solana_transaction_executor=debug,settlement_pipelines=debug,init_settlement=debug
       #  RUST_BACKTRACE: full
     commands:
     - '. "$HOME/.cargo/env"'

common-rs/src/settlements.rs

@@ -1,13 +1,15 @@
+use crate::bonds::get_bonds_for_pubkeys;
+use crate::get_validator_bonds_program;
+
+use crate::utils::get_accounts_for_pubkeys;
+
 use solana_client::nonblocking::rpc_client::RpcClient;
 use solana_sdk::pubkey::Pubkey;
-
+use std::collections::HashSet;
 use std::sync::Arc;
-
+use validator_bonds::state::bond::Bond;
 use validator_bonds::state::settlement::Settlement;
 
-use crate::get_validator_bonds_program;
-use crate::utils::get_accounts_for_pubkeys;
-
 pub async fn get_settlements(
     rpc_client: Arc<RpcClient>,
 ) -> anyhow::Result<Vec<(Pubkey, Settlement)>> {
@@ -21,3 +23,38 @@ pub async fn get_settlements_for_pubkeys(
 ) -> anyhow::Result<Vec<(Pubkey, Option<Settlement>)>> {
     get_accounts_for_pubkeys(rpc_client, pubkeys).await
 }
+
+pub async fn get_bonds_for_settlements(
+    rpc_client: Arc<RpcClient>,
+    settlements: &[(Pubkey, Settlement)],
+) -> anyhow::Result<Vec<(Pubkey, Option<Bond>)>> {
+    let bond_pubkeys = settlements
+        .iter()
+        .map(|(_, settlement)| settlement.bond)
+        .collect::<HashSet<_>>() // be unique
+        .into_iter()
+        .collect::<Vec<Pubkey>>();
+
+    let bonds = get_bonds_for_pubkeys(rpc_client, &bond_pubkeys).await?;
+
+    let settlements_bonds = settlements
+        .iter()
+        .map(|(pubkey, settlement)| {
+            bonds
+                .iter()
+                .find(|(bond_pubkey, bond)| bond_pubkey == pubkey && bond.is_some())
+                .map_or_else(
+                    || (settlement.bond, None),
+                    |(_, bond)| {
+                        if let Some(bond) = bond {
+                            (settlement.bond, Some(bond.clone()))
+                        } else {
+                            (settlement.bond, None)
+                        }
+                    },
+                )
+        })
+        .collect();
+
+    Ok(settlements_bonds)
+}

common-rs/src/stake_accounts.rs

@@ -32,6 +32,7 @@ pub async fn get_clock(rpc_client: Arc<RpcClient>) -> anyhow::Result<Clock> {
     )?)
 }
 
+/// stake account pubkey, lamports in account, stake state
 pub type CollectedStakeAccounts = Vec<(Pubkey, u64, StakeStateV2)>;
 
 pub async fn collect_stake_accounts(

packages/validator-bonds-cli/README.md

@@ -713,4 +713,4 @@ Commands:
   Alternatively, cancel the current withdrawal request and create a new one with a larger `--amount` to cover
   additional staking rewards earned by stake accounts during the delayed period until claiming is permitted.
   For example, the new withdrawal request could specify `--amount` as 5 SOLs instead of 4 SOLs for this particular example,
-  or use term `"ALL"` (`--amount ALL`) to declare the desire to withdraw everything from the bond regardless of anything.
+  or use term `"ALL"` (`--amount ALL`) to declare the desire to withdraw everything from the bond regardless of anything.

settlement-pipelines/Cargo.toml

@@ -17,6 +17,14 @@ path = "src/bin/list_claimable_epoch.rs"
 name = "claim-settlement"
 path = "src/bin/claim_settlement.rs"
 
+[[bin]]
+name = "close-settlement"
+path = "src/bin/close_settlement.rs"
+
+[[bin]]
+name = "list-settlement"
+path = "src/bin/list_settlement.rs"
+
 [dependencies]
 anchor-client = {workspace = true}
 anyhow = { workspace = true }

settlement-pipelines/__tests__/test-validator/pipelineSettlement.spec.ts

@@ -484,8 +484,9 @@ describe.skip('Cargo CLI: Pipeline Settlement', () => {
           configAccount.toBase58(),
           '--rpc-url',
           provider.connection.rpcEndpoint,
-          '-d',
-          merkleTreesDir,
+          '-s',
+          merkleTreeCollectionPath,
+          settlementCollectionPath,
           '--epoch',
           currentEpoch.toString(),
           '--fee-payer',
@@ -514,8 +515,9 @@ describe.skip('Cargo CLI: Pipeline Settlement', () => {
           configAccount.toBase58(),
           '--rpc-url',
           provider.connection.rpcEndpoint,
-          '-d',
-          merkleTreesDir,
+          '--settlement-json-files',
+          merkleTreeCollectionPath,
+          settlementCollectionPath,
           '--epoch',
           currentEpoch.toString(),
         ],

settlement-pipelines/src/arguments.rs

@@ -23,7 +23,7 @@ pub struct GlobalOpts {
         env,
         default_value = "https://api.mainnet-beta.solana.com"
     )]
-    pub rpc_url: Option<String>,
+    pub rpc_url: String,
 
     #[arg(long = "commitment", default_value = "confirmed")]
     pub commitment: CommitmentLevel,
@@ -68,15 +68,15 @@ pub struct TipPolicyOpts {
     tip_multiplier: Option<u64>,
 }
 
-pub fn load_default_keypair(s: Option<&str>) -> Result<Option<Rc<Keypair>>, anyhow::Error> {
+pub fn load_default_keypair(s: Option<&str>) -> anyhow::Result<Option<Rc<Keypair>>> {
     if s.is_none() || s.unwrap().is_empty() {
         load_keypair(DEFAULT_KEYPAIR_PATH).map_or_else(|_e| Ok(None), |keypair| Ok(Some(keypair)))
     } else {
         Ok(Some(load_keypair(s.unwrap())?))
     }
 }
 
-pub fn load_keypair(s: &str) -> Result<Rc<Keypair>, anyhow::Error> {
+pub fn load_keypair(s: &str) -> anyhow::Result<Rc<Keypair>> {
     // loading directly as the json keypair data (format [u8; 64])
     let parsed_json = parse_keypair_as_json_data(s);
     if let Ok(key_bytes) = parsed_json {
@@ -96,6 +96,21 @@ pub fn load_keypair(s: &str) -> Result<Rc<Keypair>, anyhow::Error> {
     Ok(Rc::new(k))
 }
 
+pub fn load_pubkey(s: &str) -> anyhow::Result<Pubkey> {
+    let parsed_keypair_data = parse_keypair_as_json_data(s);
+    if let Ok(keypair_data) = parsed_keypair_data {
+        if let Ok(keypair) = Keypair::from_bytes(&keypair_data) {
+            Ok(keypair.pubkey())
+        } else {
+            Err(anyhow!(
+                "Could not read pubkey from json data that seems to be a keypair"
+            ))
+        }
+    } else {
+        Pubkey::from_str(s).map_err(|e| anyhow!("Could not parse pubkey from '{}': {}", s, e))
+    }
+}
+
 fn create_clap_error(message: &str, context_value: &str) -> clap::Error {
     let mut err = clap::Error::raw(clap::error::ErrorKind::ValueValidation, message);
     err.insert(
@@ -127,11 +142,9 @@ pub fn to_tip_policy(opts: &TipPolicyOpts) -> TipPolicy {
     }
 }
 
-pub fn to_priority_fee_policy(
-    opts: &PriorityFeePolicyOpts,
-) -> solana_transaction_executor::PriorityFeePolicy {
-    let default = solana_transaction_executor::PriorityFeePolicy::default();
-    solana_transaction_executor::PriorityFeePolicy {
+pub fn to_priority_fee_policy(opts: &PriorityFeePolicyOpts) -> PriorityFeePolicy {
+    let default = PriorityFeePolicy::default();
+    PriorityFeePolicy {
         micro_lamports_per_cu_min: opts
             .micro_lamports_per_cu_min
             .unwrap_or(default.micro_lamports_per_cu_min),
@@ -145,25 +158,34 @@ pub fn to_priority_fee_policy(
 }
 
 pub struct InitializedGlobalOpts {
-    pub rpc_url: String,
-    pub fee_payer_keypair: Rc<Keypair>,
-    pub fee_payer_pubkey: Pubkey,
-    pub operator_authority_keypair: Rc<Keypair>,
+    pub fee_payer: Rc<Keypair>,
+    pub operator_authority: Rc<Keypair>,
     pub priority_fee_policy: PriorityFeePolicy,
     pub tip_policy: TipPolicy,
     pub rpc_client: Arc<RpcClient>,
     pub program: Program<Rc<DynSigner>>,
 }
 
+/// Initialize the Anchor Solana client
+pub fn get_rpc_client(global_opts: &GlobalOpts) -> anyhow::Result<(Arc<RpcClient>, String)> {
+    let rpc_url = global_opts.rpc_url.clone();
+    let anchor_cluster = Cluster::from_str(&rpc_url)
+        .map_err(|e| anyhow!("Could not parse JSON RPC url `{:?}`: {}", rpc_url, e))?;
+    let rpc_client = Arc::new(RpcClient::new_with_commitment(
+        anchor_cluster.to_string(),
+        CommitmentConfig {
+            commitment: CommitmentLevel::Confirmed,
+        },
+    ));
+    Ok((rpc_client, rpc_url))
+}
+
 pub fn init_from_opts(
     global_opts: &GlobalOpts,
     priority_fee_policy_opts: &PriorityFeePolicyOpts,
     tip_policy_opts: &TipPolicyOpts,
 ) -> anyhow::Result<InitializedGlobalOpts> {
-    // Initialize the Anchor Solana client
-    let rpc_url = global_opts.rpc_url.clone().expect("RPC URL is required");
-    let anchor_cluster = Cluster::from_str(&rpc_url)
-        .map_err(|e| anyhow!("Could not parse JSON RPC url `{:?}`: {}", rpc_url, e))?;
+    let (rpc_client, _) = get_rpc_client(global_opts)?;
 
     let default_keypair = load_default_keypair(global_opts.keypair.as_deref())?;
     let fee_payer_keypair = if let Some(fee_payer) = global_opts.fee_payer.clone() {
@@ -186,22 +208,13 @@ pub fn init_from_opts(
     let priority_fee_policy = to_priority_fee_policy(priority_fee_policy_opts);
     let tip_policy = to_tip_policy(tip_policy_opts);
 
-    let fee_payer_pubkey: Pubkey = fee_payer_keypair.pubkey();
-    let rpc_client = Arc::new(RpcClient::new_with_commitment(
-        anchor_cluster.to_string(),
-        CommitmentConfig {
-            commitment: global_opts.commitment,
-        },
-    ));
     // TODO: need to work correctly with Rc Dynsigner; need to refactor the builder executor
     let dyn_fee_payer = Rc::new(DynSigner(Arc::new(fee_payer_keypair.clone())));
     let program = get_validator_bonds_program(rpc_client.clone(), Some(dyn_fee_payer))?;
 
     Ok(InitializedGlobalOpts {
-        rpc_url,
-        fee_payer_keypair,
-        fee_payer_pubkey,
-        operator_authority_keypair,
+        fee_payer: fee_payer_keypair,
+        operator_authority: operator_authority_keypair,
         priority_fee_policy,
         tip_policy,
         rpc_client,