contributor-rewards: fail the tick instead of saving a snapshot with no leader schedule (#413)

## Summary of Changes
* The scheduler no longer writes a snapshot it cannot use: a failed leader-schedule fetch propagates instead of becoming `leader_schedule: null`, and `create_epoch_snapshot` validates before saving.
* A failed fetch was one `warn!`, then an unusable snapshot uploaded over the epoch's canonical S3 key; the tick then failed reading it back with `Missing leader schedule`, a symptom whose cause survived only in the log. #411 fixed this for the `snapshot` CLI command but not for the scheduler, which is the path that runs in production.
* Scheduler failures now log the full cause chain (`{:#}`). Plain `{}` prints only the outermost context, so the newly propagated reason would still have been dropped.
* **Security:** `EpochFinder`'s retry logs printed `reqwest`'s error verbatim, which includes the request URL. That URL carries the read endpoint's API key on mainnet-beta and journald ships to Loki, so those logs now strip it. This matters because the companion PR points mainnet-beta's reads at the keyed endpoint.
* **Behavior change:** a `--dry-run` tick that previously marked the epoch processed with an unusable snapshot now fails and retries — nothing validated on that path, since `calculate_rewards` never ran. Both deployed environments run with dry-run off.

## Testing Verification
* New `tests/test_snapshot_validate.rs` builds a `CompleteSnapshot` from the existing `testnet_snapshot.json` and `leader-schedule-epoch-89.json` fixtures: `validate()` is `Ok` on a complete snapshot, and names the right issue when the leader schedule is missing or empty.

Author: nikw9944

crates/contributor-rewards/CHANGELOG.md

@@ -7,6 +7,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+- fix(contributor-rewards): the scheduler no longer writes a snapshot it cannot use. A failed leader-schedule fetch was warned and discarded, so an unusable snapshot overwrote the epoch's canonical S3 key and the tick then failed reading it back with "Missing leader schedule". Both producers now propagate the fetch error, and the scheduler validates before saving, which also covers `--dry-run`, where nothing validated at all. Scheduler failures log the full cause chain, and every `EpochFinder` RPC error is stripped of its request URL, in the retry logs and in the error it propagates, since that URL carries the mainnet-beta read endpoint's API key into journald and Loki (malbeclabs/infra#2372)
 - fix(contributor-rewards): resolve the Solana epoch for a timestamp from real block times instead of dividing wall clock by a hardcoded 400ms slot duration. The old estimate drifted about 30k slots per day of lookback and picked the wrong epoch near a boundary, and no fixed constant survives the SIMD-0525 rollout. That epoch selects the leader schedule rewards are computed against, so the search now errors rather than returning a wrong answer: a backfill older than the endpoint's ledger retention fails on the `ingestor::demand` path instead of silently mis-estimating (malbeclabs/infra#2317)
 - fix(contributor-rewards): `snapshot` validates before writing. It warns and continues when the leader schedule cannot be fetched, but every consumer rejects a snapshot without one, so the command exited 0 having written an unusable file under the canonical name and a `snapshot` then `export-shapley` chain failed a step late. Pre-existing, but reachable now that resolving the Solana epoch depends on block-time reads (malbeclabs/infra#2317)
 - migrate to Solana 3.0: workspace `solana-*` crates and `solana-sdk` move to the 3.0 line, `solana-program-test` to 3.0.12, and the doublezero SDK git-deps repin from `client/v0.27.1` to the malbeclabs/doublezero#3830 merge revision (malbeclabs/infra#1853)

crates/contributor-rewards/src/cli/snapshot.rs

@@ -1,6 +1,6 @@
 use std::path::PathBuf;
 
-use anyhow::{Result, bail};
+use anyhow::{Context, Result, bail};
 use serde::{Deserialize, Serialize};
 use tracing::{info, warn};
 
@@ -247,27 +247,18 @@ pub async fn create_snapshot(
         fetch_data.dz_internet = internet_data;
     }
 
-    // Try to get Solana epoch and leader schedule
     let mut epoch_finder = EpochFinder::new(
         fetcher.dz_rpc_client.clone(),
         fetcher.solana_read_client.clone(),
     );
-    // fetch_leader_schedule resolves the Solana epoch itself and reports which
-    // one it used, so taking the epoch from its result avoids running the
-    // chain-verified epoch search twice over the same timestamp.
-    let leader_schedule = match epoch_finder
+    // Required: validate() below rejects a snapshot without a leader schedule.
+    // fetch_leader_schedule resolves the Solana epoch itself and reports which one it
+    // used, so taking the epoch from its result avoids running the chain-verified
+    // epoch search twice over the same timestamp.
+    let leader_schedule = epoch_finder
         .fetch_leader_schedule(fetch_epoch, fetch_data.start_us)
         .await
-    {
-        Ok(schedule) => Some(schedule),
-        Err(e) => {
-            warn!("Failed to get leader schedule: {}", e);
-            None
-        }
-    };
-    let solana_epoch = leader_schedule
-        .as_ref()
-        .map(|schedule| schedule.solana_epoch);
+        .with_context(|| format!("Failed to fetch leader schedule for DZ epoch {fetch_epoch}"))?;
 
     // Create metadata
     let metadata = SnapshotMetadata {
@@ -283,9 +274,9 @@ pub async fn create_snapshot(
     // Create complete snapshot
     let snapshot = CompleteSnapshot {
         dz_epoch: fetch_epoch,
-        solana_epoch,
+        solana_epoch: Some(leader_schedule.solana_epoch),
         fetch_data,
-        leader_schedule,
+        leader_schedule: Some(leader_schedule),
         metadata,
     };
 

crates/contributor-rewards/src/ingestor/epoch.rs

@@ -142,6 +142,30 @@ fn is_settled_block_error(err: &SolanaClientError) -> bool {
     is_block_unavailable(err) || is_block_cleaned_up(err)
 }
 
+// `reqwest::Error`, which `ClientErrorKind::Reqwest` is transparent over, prints the
+// request URL from both `Debug` and `Display`. On mainnet-beta that URL carries the
+// read endpoint's API key, and journald ships to Loki, so logging the error verbatim
+// publishes the key on every timeout and every 429.
+fn redacted(err: &SolanaClientError) -> String {
+    let text = format!("{err:?}");
+
+    match err.kind() {
+        ClientErrorKind::Reqwest(inner) => match inner.url() {
+            Some(url) => text.replace(url.as_str(), "<redacted>"),
+            None => text,
+        },
+        _ => text,
+    }
+}
+
+// Redacting only the retry logs is not enough: when the retry gives up, the client
+// error itself travels up the chain, and whoever prints that chain prints the URL.
+// Every RPC call in this module converts its error through here so the key cannot
+// leave, which means dropping the original as a source rather than wrapping it.
+fn redacted_error(err: SolanaClientError) -> anyhow::Error {
+    anyhow::Error::msg(redacted(&err))
+}
+
 /// Report whether a block at `first_block_slot` is close enough to `first_slot`,
 /// the first slot of an epoch, to date when that epoch began.
 ///
@@ -264,11 +288,13 @@ impl EpochFinder {
                 .retry(&ExponentialBuilder::default().with_jitter())
                 .notify(|err: &SolanaClientError, dur: Duration| {
                     info!(
-                        "retrying get_epoch_schedule error: {:?} with sleeping {:?}",
-                        err, dur
+                        "retrying get_epoch_schedule error: {} with sleeping {:?}",
+                        redacted(err),
+                        dur
                     )
                 })
-                .await?;
+                .await
+                .map_err(redacted_error)?;
             self.dz_schedule = Some(schedule);
         }
 
@@ -285,11 +311,13 @@ impl EpochFinder {
                 .retry(&ExponentialBuilder::default().with_jitter())
                 .notify(|err: &SolanaClientError, dur: Duration| {
                     info!(
-                        "retrying get_epoch_schedule error: {:?} with sleeping {:?}",
-                        err, dur
+                        "retrying get_epoch_schedule error: {} with sleeping {:?}",
+                        redacted(err),
+                        dur
                     )
                 })
-                .await?;
+                .await
+                .map_err(redacted_error)?;
             self.solana_schedule = Some(schedule);
         }
 
@@ -311,18 +339,18 @@ impl EpochFinder {
             .when(|err: &SolanaClientError| !is_settled_block_error(err))
             .notify(|err: &SolanaClientError, dur: Duration| {
                 info!(
-                    "retrying get_block_time error: {:?} with sleeping {:?}",
-                    err, dur
+                    "retrying get_block_time error: {} with sleeping {:?}",
+                    redacted(err),
+                    dur
                 )
             })
             .await;
 
         match block_time {
             Ok(block_time) => Ok(Some(block_time)),
             Err(err) if is_block_unavailable(&err) => Ok(None),
-            Err(err) => {
-                Err(err).with_context(|| format!("Failed to get block time for Solana slot {slot}"))
-            }
+            Err(err) => Err(redacted_error(err))
+                .with_context(|| format!("Failed to get block time for Solana slot {slot}")),
         }
     }
 
@@ -368,11 +396,13 @@ impl EpochFinder {
         .when(|err: &SolanaClientError| !is_settled_block_error(err))
         .notify(|err: &SolanaClientError, dur: Duration| {
             info!(
-                "retrying get_blocks_with_limit error: {:?} with sleeping {:?}",
-                err, dur
+                "retrying get_blocks_with_limit error: {} with sleeping {:?}",
+                redacted(err),
+                dur
             )
         })
         .await
+        .map_err(redacted_error)
         .with_context(|| format!("Failed to find the first block of Solana epoch {epoch}"))?
         .first()
         .copied()
@@ -458,9 +488,14 @@ impl EpochFinder {
         let current_slot = (|| async { self.solana_read_client.get_slot().await })
             .retry(&ExponentialBuilder::default().with_jitter())
             .notify(|err: &SolanaClientError, dur: Duration| {
-                info!("retrying get_slot error: {:?} with sleeping {:?}", err, dur)
+                info!(
+                    "retrying get_slot error: {} with sleeping {:?}",
+                    redacted(err),
+                    dur
+                )
             })
-            .await?;
+            .await
+            .map_err(redacted_error)?;
 
         let current_time_us = Utc::now().timestamp_micros() as u64;
 
@@ -576,11 +611,13 @@ impl EpochFinder {
         .retry(&ExponentialBuilder::default().with_jitter())
         .notify(|err: &SolanaClientError, dur: Duration| {
             info!(
-                "retrying get_leader_schedule error: {:?} with sleeping {:?}",
-                err, dur
+                "retrying get_leader_schedule error: {} with sleeping {:?}",
+                redacted(err),
+                dur
             )
         })
-        .await?
+        .await
+        .map_err(redacted_error)?
         .ok_or_else(|| anyhow!("No leader schedule found for Solana epoch {solana_epoch}"))?;
 
         // Convert leader schedule to map of validator -> slot count
@@ -607,6 +644,32 @@ mod tests {
 
     use super::*;
 
+    // Points a client at a closed local port so the call fails inside reqwest with the
+    // URL attached, which is the shape that leaks the API key.
+    #[tokio::test]
+    async fn test_redaction_keeps_the_url_out_of_a_propagated_error() {
+        let err = RpcClient::new("http://127.0.0.1:1/?api-key=SUPERSECRET".to_string())
+            .get_slot()
+            .await
+            .expect_err("a closed port cannot answer get_slot");
+
+        // The redaction depends on both of these, so check them rather than letting a
+        // reqwest change turn the assertions below into a vacuous pass.
+        assert!(matches!(err.kind(), ClientErrorKind::Reqwest(_)), "{err:?}");
+        let ClientErrorKind::Reqwest(inner) = err.kind() else {
+            unreachable!()
+        };
+        assert!(inner.url().is_some(), "{err:?}");
+
+        assert!(format!("{err:?}").contains("SUPERSECRET"));
+        assert!(!redacted(&err).contains("SUPERSECRET"));
+
+        // What the scheduler prints on failure, via `error!("...: {e:#}")`.
+        let propagated = redacted_error(err);
+        assert!(!format!("{propagated:#}").contains("SUPERSECRET"));
+        assert!(!format!("{propagated:?}").contains("SUPERSECRET"));
+    }
+
     #[test]
     fn test_estimate_slot_from_timestamp() {
         let current_slot = 1000000;

crates/contributor-rewards/src/scheduler/worker.rs

@@ -8,7 +8,7 @@ use std::{
     time::{Duration, Instant},
 };
 
-use anyhow::{Result, anyhow, bail, ensure};
+use anyhow::{Context, Result, anyhow, bail, ensure};
 use backon::{ExponentialBuilder, Retryable};
 use chrono::Utc;
 use doublezero_program_tools::zero_copy;
@@ -147,7 +147,7 @@ impl ScheduleWorker {
                     state.save(&self.state_file)?;
                 }
                 Err(e) => {
-                    error!("Failed to process rewards: {}", e);
+                    error!("Failed to process rewards: {e:#}");
                     state.mark_failure();
                     state.save(&self.state_file)?;
 
@@ -418,10 +418,9 @@ impl ScheduleWorker {
                     (location, path, temp_guard)
                 }
                 Err(e) => {
-                    error!(
-                        "Failed to create snapshot for epoch {}: {}",
-                        target_epoch, e
-                    );
+                    // `{:#}` prints the cause chain; plain `{}` prints only the outermost
+                    // context, dropping the reason the fetch failed.
+                    error!("Failed to create snapshot for epoch {target_epoch}: {e:#}");
                     metrics::counter!(
                         "doublezero_contributor_rewards_snapshot_failed",
                         "reason" => "creation_error"
@@ -769,28 +768,19 @@ impl ScheduleWorker {
             );
         }
 
-        // Try to fetch leader schedule (optional - warn on failure)
+        // Required: every consumer rejects a snapshot without a leader schedule.
         info!("Fetching leader schedule for epoch {}", epoch);
-        let (solana_epoch, leader_schedule) = match EpochFinder::new(
+        let leader_schedule = EpochFinder::new(
             fetcher.dz_rpc_client.clone(),
             fetcher.solana_read_client.clone(),
         )
         .fetch_leader_schedule(epoch, fetch_data.start_us)
         .await
-        {
-            Ok(schedule) => {
-                info!(
-                    "Leader schedule fetched successfully for Solana epoch {}",
-                    schedule.solana_epoch
-                );
-                (Some(schedule.solana_epoch), Some(schedule))
-            }
-            Err(e) => {
-                warn!("Failed to fetch leader schedule for epoch {}: {}", epoch, e);
-                warn!("Snapshot will be created without leader schedule");
-                (None, None)
-            }
-        };
+        .with_context(|| format!("Failed to fetch leader schedule for DZ epoch {epoch}"))?;
+        info!(
+            "Leader schedule fetched successfully for Solana epoch {}",
+            leader_schedule.solana_epoch
+        );
 
         // Create metadata
         let metadata = SnapshotMetadata {
@@ -806,12 +796,17 @@ impl ScheduleWorker {
         // Create complete snapshot
         let snapshot = CompleteSnapshot {
             dz_epoch: epoch,
-            solana_epoch,
+            solana_epoch: Some(leader_schedule.solana_epoch),
             fetch_data,
-            leader_schedule,
+            leader_schedule: Some(leader_schedule),
             metadata,
         };
 
+        // Validate before saving: storage.save writes the canonical per-epoch key, so
+        // an incomplete snapshot overwrites a good one, and a dry run never reads it
+        // back to find out.
+        snapshot.validate()?;
+
         // Save snapshot using storage abstraction (S3 or local file)
         info!(
             "Saving snapshot using {} storage",