feat(contributor-rewards): add distribute-backfill command for past epochs

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

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<DistributionSummary> {
     // 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<usize, ContributorStatus> = HashMap::new();
 
     for (leaf_index, reward_share, is_processed) in
         try_distribution_rewards_iter(&distribution, &shapley_output)?
@@ -141,20 +156,33 @@ 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,
             &shapley_output,
             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();

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<PathBuf>,
     },
+    #[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
+        #[arg(short = 'e', long, value_name = "EPOCH")]
+        dz_epoch: u64,
+
+        /// 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());
@@ -576,6 +603,7 @@ pub async fn handle(orchestrator: &Orchestrator, cmd: RewardsCommands) -> Result
                         &config.rewards_accountant_key,
                         dz_epoch_value,
                         &shapley_prefix,
+                        false,
                     )
                     .await?;
 
@@ -630,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
@@ -657,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();
 
@@ -690,6 +727,123 @@ pub async fn handle(orchestrator: &Orchestrator, cmd: RewardsCommands) -> Result
                 (false, None) => unreachable!(),
             }
 
+            Ok(())
+        }
+        RewardsCommands::DistributeBackfill {
+            dz_epoch,
+            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 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} (dry-run)");
+            } else {
+                info!("Backfilling rewards for epoch {dz_epoch}");
+            }
+
+            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,
+                &shapley_prefix,
+                true,
+            )
+            .await?;
+
+            if matches!(summary.outcome, distribute::DistributionOutcome::NotReady) {
+                info!("Distribution not ready for epoch {dz_epoch}");
+                return Ok(());
+            }
+
+            // 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,
+                )
+                .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([
+                    "contributor".to_string(),
+                    "proportion".to_string(),
+                    "reward".to_string(),
+                    "status".to_string(),
+                ]);
+                for c in &delta {
+                    table_builder.push_record([
+                        resolve_label(&c.contributor_key),
+                        format!("{:.2}%", 100.0 * c.proportion),
+                        format!("{:.1} 2Z", c.reward_tokens),
+                        match c.status {
+                            distribute::ContributorStatus::New => "new",
+                            distribute::ContributorStatus::Skipped => "skipped",
+                            distribute::ContributorStatus::Existing => unreachable!(),
+                        }
+                        .to_string(),
+                    ]);
+                }
+                let table = table_builder
+                    .build()
+                    .with(Style::psql().remove_horizontals())
+                    .to_string();
+                println!("\n{table}");
+            }
+
             Ok(())
         }
     }

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();