From bdcf7563d2dd45f3c54acdbbb4656312005c2495 Mon Sep 17 00:00:00 2001 From: Rahul Garg Date: Mon, 16 Mar 2026 16:28:14 -0400 Subject: [PATCH 1/4] feat(contributor-rewards): add distribute-backfill command for distributing past epochs --- crates/contributor-rewards/src/cli/rewards.rs | 167 ++++++++++++++++-- 1 file changed, 157 insertions(+), 10 deletions(-) diff --git a/crates/contributor-rewards/src/cli/rewards.rs b/crates/contributor-rewards/src/cli/rewards.rs index 6b18a811..e3962180 100644 --- a/crates/contributor-rewards/src/cli/rewards.rs +++ b/crates/contributor-rewards/src/cli/rewards.rs @@ -2,7 +2,10 @@ use std::path::PathBuf; use anyhow::{Result, ensure}; use clap::Subcommand; -use doublezero_solana_client_tools::rpc::SolanaConnection; +use doublezero_solana_client_tools::{ + payer::Wallet, + rpc::{DoubleZeroLedgerConnection, SolanaConnection}, +}; use doublezero_solana_sdk::revenue_distribution::fetch::{ SolConversionState, try_fetch_config, try_fetch_distribution, }; @@ -14,7 +17,10 @@ use tabled::{builder::Builder as TableBuilder, settings::Style}; use tracing::{info, warn}; use crate::{ - calculator::{ledger_operations::WriteResult, orchestrator::Orchestrator}, + calculator::{ + distribute, keypair_loader::load_keypair, ledger_operations::WriteResult, + orchestrator::Orchestrator, + }, cli::snapshot::CompleteSnapshot, }; @@ -327,6 +333,35 @@ pub enum RewardsCommands { )] keypair: Option, }, + #[command( + about = "Re-distribute rewards for a past epoch (backfill missed contributors)", + long_about = "Ad-hoc operator tool for re-distributing rewards for a past epoch where some \ + contributors were skipped (e.g., because their ContributorRewards account didn't exist yet). \ + No Slack notifications are sent. Script in a loop for multiple epochs:\n\n\ + for e in 40..50; do distribute-backfill -e $e -k keypair.json; done", + after_help = r#"Examples: + # Backfill a specific epoch + distribute-backfill -e 42 -k keypair.json + + # Simulate backfill without sending transactions + distribute-backfill -e 42 -k keypair.json --dry-run"# + )] + DistributeBackfill { + /// DZ epoch to backfill (defaults to latest distributable epoch) + #[arg(short = 'e', long, value_name = "EPOCH")] + dz_epoch: Option, + + /// Simulate transactions without sending + #[arg(long)] + dry_run: bool, + + /// Path to keypair file for signing transactions. + /// Always required because even --dry-run simulates signed transactions + /// against the RPC. Use `distribute-rewards --dry-run -e N` for a + /// lightweight readiness check that needs no keypair. + #[arg(short = 'k', long, value_name = "FILE")] + keypair: PathBuf, + }, } /// Handle rewards commands @@ -343,8 +378,6 @@ pub async fn handle(orchestrator: &Orchestrator, cmd: RewardsCommands) -> Result skip_merkle_root, slack_notify, } => { - use tracing::warn; - // Construct WriteConfig from CLI flags let write_config = crate::calculator::WriteConfig::from_flags( skip_device_telemetry, @@ -547,12 +580,6 @@ pub async fn handle(orchestrator: &Orchestrator, cmd: RewardsCommands) -> Result } // Mode 2 & 3: Simulate or Execute (keypair present) (dry_run, Some(keypair_path)) => { - use doublezero_solana_client_tools::{ - payer::Wallet, rpc::DoubleZeroLedgerConnection, - }; - - use crate::calculator::{distribute, keypair_loader::load_keypair}; - let signer = load_keypair(&Some(keypair_path))?; let dz_connection = DoubleZeroLedgerConnection::new(orchestrator.settings.rpc.dz_url.clone()); @@ -690,6 +717,126 @@ pub async fn handle(orchestrator: &Orchestrator, cmd: RewardsCommands) -> Result (false, None) => unreachable!(), } + Ok(()) + } + RewardsCommands::DistributeBackfill { + dz_epoch, + dry_run, + keypair, + } => { + let connection = + SolanaConnection::new(orchestrator.settings.rpc.solana_write_url.clone()); + + let (_, config) = try_fetch_config(&connection).await?; + + let dz_epoch_value = match dz_epoch { + Some(epoch) => { + info!("Will backfill for provided dz_epoch: {epoch}"); + epoch + } + None => { + let sol_conversion_state = SolConversionState::try_fetch(&connection).await?; + let next_sweep = sol_conversion_state + .journal + .1 + .next_dz_epoch_to_sweep_tokens + .value(); + ensure!(next_sweep > 0, "No epochs have been swept yet"); + let dist_epoch = next_sweep - 1; + info!("Will backfill for dz_epoch: {dist_epoch}, next_sweep: {next_sweep}"); + dist_epoch + } + }; + + let signer = load_keypair(&Some(keypair))?; + let dz_connection = + DoubleZeroLedgerConnection::new(orchestrator.settings.rpc.dz_url.clone()); + + let wallet = Wallet { + connection, + signer, + compute_unit_price_ix: None, + verbose: false, + fee_payer: None, + dry_run, + }; + + if dry_run { + info!("Backfilling rewards for epoch {dz_epoch_value} (dry-run)"); + } else { + info!("Backfilling rewards for epoch {dz_epoch_value}"); + } + + let shapley_prefix = orchestrator.settings.get_contributor_rewards_prefix(); + + let summary = distribute::try_distribute_epoch_rewards( + &wallet, + &dz_connection, + &config.rewards_accountant_key, + dz_epoch_value, + &shapley_prefix, + ) + .await?; + + match &summary.outcome { + distribute::DistributionOutcome::Complete { total_contributors } => { + info!( + "Backfill epoch {dz_epoch_value} complete: {total_contributors}/{total_contributors} distributed" + ); + } + distribute::DistributionOutcome::PartiallyComplete { + total_contributors, + distributed, + skipped, + } => { + info!( + "Backfill epoch {dz_epoch_value} partially complete: {distributed}/{total_contributors} distributed, {skipped} skipped (missing ContributorRewards accounts)" + ); + } + distribute::DistributionOutcome::NotReady => { + info!("Distribution not ready for epoch {dz_epoch_value}"); + } + } + + // Print per-contributor rewards table (no Slack notification). + if !summary.contributors.is_empty() { + let labels = crate::calculator::ledger_operations::try_fetch_contributor_labels( + &dz_connection, + &orchestrator.settings.programs.serviceability_program_id, + ) + .await + .unwrap_or_default(); + + let resolve_label = |key: &solana_sdk::pubkey::Pubkey| { + labels.get(key).cloned().unwrap_or_else(|| key.to_string()) + }; + + let mut table_builder = TableBuilder::default(); + table_builder.push_record([ + "dz_epoch".to_string(), + "index".to_string(), + "contributor".to_string(), + "proportion".to_string(), + "reward".to_string(), + "distributed".to_string(), + ]); + for c in &summary.contributors { + table_builder.push_record([ + summary.dz_epoch.to_string(), + c.index.to_string(), + resolve_label(&c.contributor_key), + format!("{:.2}%", 100.0 * c.proportion), + format!("{:.1} 2Z", c.reward_tokens), + if c.distributed { "yes" } else { "no" }.to_string(), + ]); + } + let table = table_builder + .build() + .with(Style::psql().remove_horizontals()) + .to_string(); + println!("\n{table}"); + } + Ok(()) } } From 7912f033296a83d9718b7b5b59ba3891cfadda53 Mon Sep 17 00:00:00 2001 From: Rahul Garg Date: Mon, 16 Mar 2026 16:42:22 -0400 Subject: [PATCH 2/4] chore: bump CHANGELOG --- crates/contributor-rewards/CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/crates/contributor-rewards/CHANGELOG.md b/crates/contributor-rewards/CHANGELOG.md index 231950d1..8a56a8c6 100644 --- a/crates/contributor-rewards/CHANGELOG.md +++ b/crates/contributor-rewards/CHANGELOG.md @@ -7,6 +7,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +- feat(contributor-rewards): add distribute-backfill command for past epochs ([#292](https://github.com/doublezerofoundation/doublezero-offchain/pull/292)) - feat(contributor-rewards): add support for distribution slack notifications and other minor cleanups ([#285](https://github.com/doublezerofoundation/doublezero-offchain/pull/285)) ## [0.4.3](https://github.com/doublezerofoundation/doublezero-offchain/releases/tag/contributor-rewards%2Fv0.4.3) - 2026-03-04 From 7eeac4ef87e687517366442c182a3c1826ad5446 Mon Sep 17 00:00:00 2001 From: Rahul Garg Date: Tue, 24 Mar 2026 22:50:32 +0400 Subject: [PATCH 3/4] fix(contributor-rewards): make epoch mandatory for distribute-backfill cmd --- crates/contributor-rewards/src/cli/rewards.rs | 37 +++++-------------- 1 file changed, 10 insertions(+), 27 deletions(-) diff --git a/crates/contributor-rewards/src/cli/rewards.rs b/crates/contributor-rewards/src/cli/rewards.rs index e3962180..6197b33f 100644 --- a/crates/contributor-rewards/src/cli/rewards.rs +++ b/crates/contributor-rewards/src/cli/rewards.rs @@ -347,9 +347,9 @@ pub enum RewardsCommands { distribute-backfill -e 42 -k keypair.json --dry-run"# )] DistributeBackfill { - /// DZ epoch to backfill (defaults to latest distributable epoch) + /// DZ epoch to backfill #[arg(short = 'e', long, value_name = "EPOCH")] - dz_epoch: Option, + dz_epoch: u64, /// Simulate transactions without sending #[arg(long)] @@ -724,30 +724,13 @@ pub async fn handle(orchestrator: &Orchestrator, cmd: RewardsCommands) -> Result dry_run, keypair, } => { + info!("Will backfill for dz_epoch: {dz_epoch}"); + let connection = SolanaConnection::new(orchestrator.settings.rpc.solana_write_url.clone()); let (_, config) = try_fetch_config(&connection).await?; - let dz_epoch_value = match dz_epoch { - Some(epoch) => { - info!("Will backfill for provided dz_epoch: {epoch}"); - epoch - } - None => { - let sol_conversion_state = SolConversionState::try_fetch(&connection).await?; - let next_sweep = sol_conversion_state - .journal - .1 - .next_dz_epoch_to_sweep_tokens - .value(); - ensure!(next_sweep > 0, "No epochs have been swept yet"); - let dist_epoch = next_sweep - 1; - info!("Will backfill for dz_epoch: {dist_epoch}, next_sweep: {next_sweep}"); - dist_epoch - } - }; - let signer = load_keypair(&Some(keypair))?; let dz_connection = DoubleZeroLedgerConnection::new(orchestrator.settings.rpc.dz_url.clone()); @@ -762,9 +745,9 @@ pub async fn handle(orchestrator: &Orchestrator, cmd: RewardsCommands) -> Result }; if dry_run { - info!("Backfilling rewards for epoch {dz_epoch_value} (dry-run)"); + info!("Backfilling rewards for epoch {dz_epoch} (dry-run)"); } else { - info!("Backfilling rewards for epoch {dz_epoch_value}"); + info!("Backfilling rewards for epoch {dz_epoch}"); } let shapley_prefix = orchestrator.settings.get_contributor_rewards_prefix(); @@ -773,7 +756,7 @@ pub async fn handle(orchestrator: &Orchestrator, cmd: RewardsCommands) -> Result &wallet, &dz_connection, &config.rewards_accountant_key, - dz_epoch_value, + dz_epoch, &shapley_prefix, ) .await?; @@ -781,7 +764,7 @@ pub async fn handle(orchestrator: &Orchestrator, cmd: RewardsCommands) -> Result match &summary.outcome { distribute::DistributionOutcome::Complete { total_contributors } => { info!( - "Backfill epoch {dz_epoch_value} complete: {total_contributors}/{total_contributors} distributed" + "Backfill epoch {dz_epoch} complete: {total_contributors}/{total_contributors} distributed" ); } distribute::DistributionOutcome::PartiallyComplete { @@ -790,11 +773,11 @@ pub async fn handle(orchestrator: &Orchestrator, cmd: RewardsCommands) -> Result skipped, } => { info!( - "Backfill epoch {dz_epoch_value} partially complete: {distributed}/{total_contributors} distributed, {skipped} skipped (missing ContributorRewards accounts)" + "Backfill epoch {dz_epoch} partially complete: {distributed}/{total_contributors} distributed, {skipped} skipped (missing ContributorRewards accounts)" ); } distribute::DistributionOutcome::NotReady => { - info!("Distribution not ready for epoch {dz_epoch_value}"); + info!("Distribution not ready for epoch {dz_epoch}"); } } From 0dede252602d0fc70a1c7211af18c6ce835b3eed Mon Sep 17 00:00:00 2001 From: Rahul Garg Date: Wed, 25 Mar 2026 00:54:05 +0400 Subject: [PATCH 4/4] fix(contributor-rewards): show only delta table --- .../src/calculator/distribute.rs | 45 ++++++++-- crates/contributor-rewards/src/cli/rewards.rs | 84 ++++++++++++------- .../src/scheduler/worker.rs | 8 +- 3 files changed, 101 insertions(+), 36 deletions(-) diff --git a/crates/contributor-rewards/src/calculator/distribute.rs b/crates/contributor-rewards/src/calculator/distribute.rs index 9b364dcf..c186d746 100644 --- a/crates/contributor-rewards/src/calculator/distribute.rs +++ b/crates/contributor-rewards/src/calculator/distribute.rs @@ -1,3 +1,5 @@ +use std::collections::HashMap; + use anyhow::{Result, ensure}; use doublezero_solana_client_tools::{ account::zero_copy::ZeroCopyAccountOwnedData, @@ -43,6 +45,17 @@ pub enum DistributionOutcome { }, } +/// Status of a contributor in a distribution attempt. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ContributorStatus { + /// Already distributed before this run. + Existing, + /// Newly distributed by this run. + New, + /// Attempted but skipped (missing account, simulation failure, etc.). + Skipped, +} + /// Per-contributor result from a distribution attempt. #[derive(Debug)] pub struct ContributorDistributionResult { @@ -52,8 +65,8 @@ pub struct ContributorDistributionResult { pub proportion: f64, /// Human-readable 2Z amount (already divided by decimals) pub reward_tokens: f64, - /// Whether this leaf is processed (includes both pre-existing and newly distributed) - pub distributed: bool, + /// Outcome for this contributor in the current run. + pub status: ContributorStatus, } /// Full summary of a distribution attempt. @@ -71,6 +84,7 @@ pub async fn try_distribute_epoch_rewards( rewards_accountant_key: &Pubkey, dz_epoch_value: u64, shapley_prefix: &[u8], + skip_failures: bool, ) -> Result { // Fetch the distribution for this epoch. let (_, distribution) = try_fetch_distribution(&wallet.connection, dz_epoch_value).await?; @@ -128,6 +142,7 @@ pub async fn try_distribute_epoch_rewards( let mut distributed_count = 0u32; let mut skipped_count = 0u32; + let mut leaf_outcomes: HashMap = HashMap::new(); for (leaf_index, reward_share, is_processed) in try_distribution_rewards_iter(&distribution, &shapley_output)? @@ -141,7 +156,7 @@ pub async fn try_distribute_epoch_rewards( dz_epoch_value, leaf_index, reward_share.contributor_key ); - let was_distributed = try_distribute_contributor_rewards( + let was_distributed = match try_distribute_contributor_rewards( wallet, &dz_mint_key, &distribution, @@ -149,12 +164,25 @@ pub async fn try_distribute_epoch_rewards( leaf_index, reward_share, ) - .await?; + .await + { + Ok(distributed) => distributed, + Err(e) if skip_failures => { + warn!( + "Skipping epoch {dz_epoch_value} leaf {leaf_index}, contributor {}: {e}", + reward_share.contributor_key + ); + false + } + Err(e) => return Err(e), + }; if was_distributed { distributed_count += 1; + leaf_outcomes.insert(leaf_index, ContributorStatus::New); } else { skipped_count += 1; + leaf_outcomes.insert(leaf_index, ContributorStatus::Skipped); } } @@ -190,7 +218,14 @@ pub async fn try_distribute_epoch_rewards( contributor_key: reward_share.contributor_key, proportion, reward_tokens, - distributed: is_processed, + status: if is_processed { + ContributorStatus::Existing + } else { + leaf_outcomes + .get(&index) + .copied() + .unwrap_or(ContributorStatus::Skipped) + }, } }) .collect(); diff --git a/crates/contributor-rewards/src/cli/rewards.rs b/crates/contributor-rewards/src/cli/rewards.rs index 6197b33f..58be839a 100644 --- a/crates/contributor-rewards/src/cli/rewards.rs +++ b/crates/contributor-rewards/src/cli/rewards.rs @@ -603,6 +603,7 @@ pub async fn handle(orchestrator: &Orchestrator, cmd: RewardsCommands) -> Result &config.rewards_accountant_key, dz_epoch_value, &shapley_prefix, + false, ) .await?; @@ -657,7 +658,12 @@ pub async fn handle(orchestrator: &Orchestrator, cmd: RewardsCommands) -> Result resolve_label(&c.contributor_key), format!("{:.2}%", 100.0 * c.proportion), format!("{:.1} 2Z", c.reward_tokens), - if c.distributed { "yes" } else { "no" }.to_string(), + match c.status { + distribute::ContributorStatus::Existing + | distribute::ContributorStatus::New => "yes", + distribute::ContributorStatus::Skipped => "no", + } + .to_string(), ]); } let table = table_builder @@ -684,8 +690,12 @@ pub async fn handle(orchestrator: &Orchestrator, cmd: RewardsCommands) -> Result contributor: resolve_label(&c.contributor_key), proportion: format!("{:.2}%", 100.0 * c.proportion), reward: format!("{:.1} 2Z", c.reward_tokens), - distributed: if c.distributed { "yes" } else { "no" } - .to_string(), + distributed: match c.status { + distribute::ContributorStatus::Existing + | distribute::ContributorStatus::New => "yes", + distribute::ContributorStatus::Skipped => "no", + } + .to_string(), }) .collect(); @@ -758,31 +768,44 @@ pub async fn handle(orchestrator: &Orchestrator, cmd: RewardsCommands) -> Result &config.rewards_accountant_key, dz_epoch, &shapley_prefix, + true, ) .await?; - match &summary.outcome { - distribute::DistributionOutcome::Complete { total_contributors } => { - info!( - "Backfill epoch {dz_epoch} complete: {total_contributors}/{total_contributors} distributed" - ); - } - distribute::DistributionOutcome::PartiallyComplete { - total_contributors, - distributed, - skipped, - } => { - info!( - "Backfill epoch {dz_epoch} partially complete: {distributed}/{total_contributors} distributed, {skipped} skipped (missing ContributorRewards accounts)" - ); - } - distribute::DistributionOutcome::NotReady => { - info!("Distribution not ready for epoch {dz_epoch}"); - } + if matches!(summary.outcome, distribute::DistributionOutcome::NotReady) { + info!("Distribution not ready for epoch {dz_epoch}"); + return Ok(()); } - // Print per-contributor rewards table (no Slack notification). - if !summary.contributors.is_empty() { + // Compute counts by status. + let new_count = summary + .contributors + .iter() + .filter(|c| c.status == distribute::ContributorStatus::New) + .count(); + let existing_count = summary + .contributors + .iter() + .filter(|c| c.status == distribute::ContributorStatus::Existing) + .count(); + let skipped_count = summary + .contributors + .iter() + .filter(|c| c.status == distribute::ContributorStatus::Skipped) + .count(); + + info!( + "Backfill epoch {dz_epoch}: {new_count} new, {existing_count} existing, {skipped_count} skipped" + ); + + // Show only the delta (new + skipped), not existing distributions. + let delta: Vec<_> = summary + .contributors + .iter() + .filter(|c| c.status != distribute::ContributorStatus::Existing) + .collect(); + + if !delta.is_empty() { let labels = crate::calculator::ledger_operations::try_fetch_contributor_labels( &dz_connection, &orchestrator.settings.programs.serviceability_program_id, @@ -796,21 +819,22 @@ pub async fn handle(orchestrator: &Orchestrator, cmd: RewardsCommands) -> Result let mut table_builder = TableBuilder::default(); table_builder.push_record([ - "dz_epoch".to_string(), - "index".to_string(), "contributor".to_string(), "proportion".to_string(), "reward".to_string(), - "distributed".to_string(), + "status".to_string(), ]); - for c in &summary.contributors { + for c in &delta { table_builder.push_record([ - summary.dz_epoch.to_string(), - c.index.to_string(), resolve_label(&c.contributor_key), format!("{:.2}%", 100.0 * c.proportion), format!("{:.1} 2Z", c.reward_tokens), - if c.distributed { "yes" } else { "no" }.to_string(), + match c.status { + distribute::ContributorStatus::New => "new", + distribute::ContributorStatus::Skipped => "skipped", + distribute::ContributorStatus::Existing => unreachable!(), + } + .to_string(), ]); } let table = table_builder diff --git a/crates/contributor-rewards/src/scheduler/worker.rs b/crates/contributor-rewards/src/scheduler/worker.rs index fbe16a18..1598291d 100644 --- a/crates/contributor-rewards/src/scheduler/worker.rs +++ b/crates/contributor-rewards/src/scheduler/worker.rs @@ -299,6 +299,7 @@ impl ScheduleWorker { rewards_accountant_key, dz_epoch_value, &shapley_prefix, + false, ) .await } @@ -347,7 +348,12 @@ impl ScheduleWorker { contributor, proportion: format!("{:.2}%", 100.0 * c.proportion), reward: format!("{:.1} 2Z", c.reward_tokens), - distributed: if c.distributed { "yes" } else { "no" }.to_string(), + distributed: match c.status { + distribute::ContributorStatus::Existing + | distribute::ContributorStatus::New => "yes", + distribute::ContributorStatus::Skipped => "no", + } + .to_string(), } }) .collect();