Merge branch 'main' into feat/add-validator-submitter-ci-job

biome.json

@@ -17,7 +17,8 @@
       "!**/html/**/*",
       "!**/moonwall/contracts/out/**/*",
       "!**/contracts/out/**/*",
-      "!**/contracts/deployments/state-diff.checksum"
+      "!**/contracts/deployments/state-diff.checksum",
+      "!**/bun.lock"
     ],
     "maxSize": 3000000
   },

operator/.dockerignore

@@ -50,4 +50,3 @@ examples/
 Cargo.lock.old
 *.toml.old
 *.lock.old
-**/target/

operator/Dockerfile

@@ -55,7 +55,7 @@ COPY --from=builder \
 RUN useradd -m -u 1001 -U -s /bin/sh -d /datahaven datahaven && \
     mkdir -p /datahaven/.local/share /data && \
     chown -R datahaven:datahaven /data && \
-    ln -s /data /datahaven/.local/share/datahaven
+    ln -s /data /datahaven/.local/share/datahaven-node
 
 USER datahaven
 

operator/pallets/external-validators-rewards/src/benchmarking.rs

@@ -21,9 +21,9 @@ use super::*;
 #[allow(unused)]
 use crate::Pallet as ExternalValidatorsRewards;
 use {
-    crate::{types::BenchmarkHelper, OnEraEnd},
+    crate::types::BenchmarkHelper,
     frame_benchmarking::{account, v2::*, BenchmarkError},
-    frame_support::traits::Currency,
+    frame_support::traits::{Currency, EnsureOrigin},
     sp_std::prelude::*,
 };
 
@@ -43,6 +43,11 @@ fn create_funded_user<T: Config + pallet_balances::Config>(
     user
 }
 
+/// Helper: insert a single entry into the ring buffer at slot 0.
+fn push_unsent_entry<T: Config>(era_index: u32, timestamp: u32, inflation: u128) {
+    ExternalValidatorsRewards::<T>::unsent_queue_push((era_index, timestamp, inflation));
+}
+
 #[allow(clippy::multiple_bound_locations)]
 #[benchmarks(where T: pallet_balances::Config)]
 mod benchmarks {
@@ -72,6 +77,106 @@ mod benchmarks {
         Ok(())
     }
 
+    /// Helper to populate reward points for an era with 1000 validators.
+    fn setup_era_reward_points<T: Config + pallet_balances::Config>(era_index: u32) {
+        let mut era_reward_points = EraRewardPoints::default();
+        era_reward_points.total = 20 * 1000;
+
+        for i in 0..1000 {
+            let account_id = create_funded_user::<T>("candidate", i, 100);
+            era_reward_points.individual.insert(account_id, 20);
+        }
+
+        <RewardPointsForEra<T>>::insert(era_index, era_reward_points);
+    }
+
+    // on_initialize: unsent queue is empty (2 reads for head+tail)
+    #[benchmark]
+    fn process_unsent_reward_eras_empty() -> Result<(), BenchmarkError> {
+        // Ensure queue is empty (default state: head == tail == 0)
+        assert!(ExternalValidatorsRewards::<T>::unsent_queue_is_empty());
+
+        #[block]
+        {
+            ExternalValidatorsRewards::<T>::process_unsent_reward_eras();
+        }
+
+        Ok(())
+    }
+
+    // on_initialize: oldest entry has pruned reward points
+    #[benchmark]
+    fn process_unsent_reward_eras_expired() -> Result<(), BenchmarkError> {
+        // Push an entry whose reward points do NOT exist in storage
+        push_unsent_entry::<T>(999, 0, 42);
+
+        #[block]
+        {
+            ExternalValidatorsRewards::<T>::process_unsent_reward_eras();
+        }
+
+        // Entry should have been removed
+        assert!(ExternalValidatorsRewards::<T>::unsent_queue_is_empty());
+
+        Ok(())
+    }
+
+    // on_initialize: oldest entry retried successfully
+    #[benchmark]
+    fn process_unsent_reward_eras_success() -> Result<(), BenchmarkError> {
+        frame_system::Pallet::<T>::set_block_number(0u32.into());
+        T::BenchmarkHelper::setup();
+        setup_era_reward_points::<T>(1);
+
+        push_unsent_entry::<T>(1, 0, 42);
+
+        #[block]
+        {
+            ExternalValidatorsRewards::<T>::process_unsent_reward_eras();
+        }
+
+        assert!(ExternalValidatorsRewards::<T>::unsent_queue_is_empty());
+
+        Ok(())
+    }
+
+    // Use success weight as upper bound for the failed path
+    #[benchmark]
+    fn process_unsent_reward_eras_failed() -> Result<(), BenchmarkError> {
+        frame_system::Pallet::<T>::set_block_number(0u32.into());
+        T::BenchmarkHelper::setup();
+        setup_era_reward_points::<T>(1);
+
+        push_unsent_entry::<T>(1, 0, 42);
+
+        #[block]
+        {
+            ExternalValidatorsRewards::<T>::process_unsent_reward_eras();
+        }
+
+        Ok(())
+    }
+
+    // Governance extrinsic: retry a specific unsent era
+    #[benchmark]
+    fn retry_unsent_reward_era() -> Result<(), BenchmarkError> {
+        frame_system::Pallet::<T>::set_block_number(0u32.into());
+        T::BenchmarkHelper::setup();
+        setup_era_reward_points::<T>(1);
+
+        push_unsent_entry::<T>(1, 0, 42);
+
+        let origin =
+            T::GovernanceOrigin::try_successful_origin().map_err(|_| BenchmarkError::Weightless)?;
+
+        #[extrinsic_call]
+        _(origin as T::RuntimeOrigin, 1u32);
+
+        assert!(ExternalValidatorsRewards::<T>::unsent_queue_is_empty());
+
+        Ok(())
+    }
+
     impl_benchmark_test_suite!(
         ExternalValidatorsRewards,
         crate::mock::new_test_ext(),

operator/pallets/external-validators-rewards/src/lib.rs

@@ -66,13 +66,13 @@ pub mod pallet {
 
     pub use crate::weights::WeightInfo;
     use {
-        super::*, frame_support::pallet_prelude::*,
+        super::*, frame_support::pallet_prelude::*, frame_system::pallet_prelude::OriginFor,
         pallet_external_validators::traits::EraIndexProvider, sp_runtime::Saturating,
         sp_std::collections::btree_map::BTreeMap,
     };
 
     /// The current storage version.
-    const STORAGE_VERSION: StorageVersion = StorageVersion::new(0);
+    const STORAGE_VERSION: StorageVersion = StorageVersion::new(1);
 
     pub type RewardPoints = u32;
     pub type EraIndex = u32;
@@ -168,6 +168,9 @@ pub mod pallet {
         /// Hook for minting inflation tokens.
         type HandleInflation: HandleInflation<Self::AccountId>;
 
+        /// Origin for governance calls (e.g., retrying unsent reward messages).
+        type GovernanceOrigin: EnsureOrigin<Self::RuntimeOrigin>;
+
         #[cfg(feature = "runtime-benchmarks")]
         type BenchmarkHelper: types::BenchmarkHelper;
     }
@@ -175,6 +178,62 @@ pub mod pallet {
     #[pallet::storage_version(STORAGE_VERSION)]
     pub struct Pallet<T>(_);
 
+    #[pallet::hooks]
+    impl<T: Config> Hooks<frame_system::pallet_prelude::BlockNumberFor<T>> for Pallet<T> {
+        fn on_initialize(_n: frame_system::pallet_prelude::BlockNumberFor<T>) -> Weight {
+            Self::process_unsent_reward_eras()
+        }
+    }
+
+    #[pallet::call]
+    impl<T: Config> Pallet<T> {
+        /// Governance escape hatch: manually retry sending a rewards message for
+        /// an era that is stuck in the unsent queue.
+        #[pallet::call_index(0)]
+        #[pallet::weight(T::WeightInfo::retry_unsent_reward_era())]
+        pub fn retry_unsent_reward_era(
+            origin: OriginFor<T>,
+            era_index: EraIndex,
+        ) -> DispatchResult {
+            T::GovernanceOrigin::ensure_origin(origin)?;
+
+            // Scan the ring buffer for the requested era
+            let head = UnsentRewardHead::<T>::get();
+            let tail = UnsentRewardTail::<T>::get();
+            let mut found = None;
+            let mut slot = head;
+            while slot != tail {
+                if let Some(entry @ (idx, _, _)) = UnsentRewardEra::<T>::get(slot) {
+                    if idx == era_index {
+                        found = Some((slot, entry));
+                        break;
+                    }
+                }
+                slot = (slot + 1) % UNSENT_QUEUE_CAPACITY;
+            }
+            let (slot, (_, timestamp, inflation)) = found.ok_or(Error::<T>::EraNotInUnsentQueue)?;
+
+            let reward_points = RewardPointsForEra::<T>::get(era_index);
+            let info = reward_points
+                .generate_era_rewards_info(era_index, inflation, timestamp)
+                .ok_or(Error::<T>::RewardPointsPruned)?;
+
+            let message_id =
+                Self::send_rewards_message(&info).ok_or(Error::<T>::MessageSendFailed)?;
+
+            Self::unsent_queue_remove_slot(slot);
+
+            Self::deposit_event(Event::RewardsMessageRetried {
+                message_id,
+                era_index,
+                total_points: info.total_points,
+                inflation_amount: inflation,
+            });
+
+            Ok(())
+        }
+    }
+
     #[pallet::event]
     #[pallet::generate_deposit(pub(super) fn deposit_event)]
     pub enum Event<T: Config> {
@@ -185,6 +244,29 @@ pub mod pallet {
             total_points: u128,
             inflation_amount: u128,
         },
+        /// The rewards message failed to send; era queued for retry.
+        RewardsMessageSendFailed { era_index: EraIndex },
+        /// A previously failed rewards message was retried and sent successfully.
+        RewardsMessageRetried {
+            message_id: H256,
+            era_index: EraIndex,
+            total_points: u128,
+            inflation_amount: u128,
+        },
+        /// An unsent era was dropped because its reward points have been pruned.
+        UnsentEraExpired { era_index: EraIndex },
+        /// The unsent queue is full; this era could not be enqueued for retry.
+        UnsentQueueFull { era_index: EraIndex },
+    }
+
+    #[pallet::error]
+    pub enum Error<T> {
+        /// The specified era is not in the unsent queue.
+        EraNotInUnsentQueue,
+        /// Reward points for the era have been pruned from storage.
+        RewardPointsPruned,
+        /// The message delivery still failed on retry.
+        MessageSendFailed,
     }
 
     /// Keep tracks of distributed points per validator and total.
@@ -200,7 +282,7 @@ pub mod pallet {
         ///  - individual_points: (address, points) tuples for each validator.
         ///  - inflation_amount: total inflation tokens to distribute.
         ///  - era_start_timestamp: timestamp when the era started (seconds since Unix epoch).
-        pub fn generate_era_rewards_utils(
+        pub fn generate_era_rewards_info(
             &self,
             era_index: EraIndex,
             inflation_amount: u128,
@@ -260,6 +342,33 @@ pub mod pallet {
     pub type BlocksProducedInEra<T: Config> =
         StorageMap<_, Twox64Concat, EraIndex, u32, ValueQuery>;
 
+    /// Maximum number of unsent reward entries in the ring buffer.
+    pub const UNSENT_QUEUE_CAPACITY: u32 = 64;
+
+    /// Ring buffer of eras whose rewards messages failed to send.
+    /// Each slot stores (era_index, era_start_timestamp, scaled_inflation).
+    /// Keyed by slot index [0, UNSENT_QUEUE_CAPACITY).
+    #[pallet::storage]
+    pub type UnsentRewardEra<T: Config> = StorageMap<
+        _,
+        Twox64Concat,
+        u32,
+        (
+            EraIndex,
+            /* era_start_timestamp */ u32,
+            /* scaled_inflation */ u128,
+        ),
+    >;
+
+    /// Ring buffer head: next slot to be processed by `on_initialize`.
+    #[pallet::storage]
+    pub type UnsentRewardHead<T: Config> = StorageValue<_, u32, ValueQuery>;
+
+    /// Ring buffer tail: next slot to write a new entry into.
+    /// When head == tail the buffer is empty.
+    #[pallet::storage]
+    pub type UnsentRewardTail<T: Config> = StorageValue<_, u32, ValueQuery>;
+
     impl<T: Config> Pallet<T> {
         /// Reward validators. Does not check if the validators are valid, caller needs to make sure of that.
         pub fn reward_by_ids(points: impl IntoIterator<Item = (T::AccountId, RewardPoints)>) {
@@ -276,8 +385,8 @@ pub mod pallet {
 
         /// Helper to build, validate and deliver an outbound message.
         /// Logs any error and returns None on failure.
-        fn send_rewards_message(utils: &EraRewardsUtils) -> Option<H256> {
-            let outbound = T::SendMessage::build(utils).or_else(|| {
+        fn send_rewards_message(info: &EraRewardsUtils) -> Option<H256> {
+            let outbound = T::SendMessage::build(info).or_else(|| {
                 log::error!(target: "ext_validators_rewards", "Failed to build outbound message");
                 None
             })?;
@@ -303,6 +412,147 @@ pub mod pallet {
                 .ok()
         }
 
+        // ── Ring-buffer helpers ──────────────────────────────────────────
+
+        /// Returns true when the ring buffer is empty (head == tail).
+        #[allow(dead_code)]
+        pub(crate) fn unsent_queue_is_empty() -> bool {
+            UnsentRewardHead::<T>::get() == UnsentRewardTail::<T>::get()
+        }
+
+        /// Number of entries currently in the ring buffer.
+        #[allow(dead_code)]
+        pub(crate) fn unsent_queue_len() -> u32 {
+            let head = UnsentRewardHead::<T>::get();
+            let tail = UnsentRewardTail::<T>::get();
+            tail.wrapping_sub(head) % UNSENT_QUEUE_CAPACITY
+        }
+
+        /// Push a new entry into the ring buffer.
+        /// Returns `true` on success, `false` if the buffer is full.
+        pub(crate) fn unsent_queue_push(entry: (EraIndex, u32, u128)) -> bool {
+            let head = UnsentRewardHead::<T>::get();
+            let tail = UnsentRewardTail::<T>::get();
+            let next_tail = (tail + 1) % UNSENT_QUEUE_CAPACITY;
+            if next_tail == head {
+                // Buffer full
+                return false;
+            }
+            UnsentRewardEra::<T>::insert(tail, entry);
+            UnsentRewardTail::<T>::put(next_tail);
+            true
+        }
+
+        /// Remove the entry at a given slot and compact the buffer by shifting
+        /// subsequent entries back. Used by the extrinsic and `on_era_start`.
+        fn unsent_queue_remove_slot(slot: u32) {
+            let tail = UnsentRewardTail::<T>::get();
+            // Shift entries after `slot` backward to fill the gap
+            let mut cur = slot;
+            loop {
+                let next = (cur + 1) % UNSENT_QUEUE_CAPACITY;
+                if next == tail {
+                    break;
+                }
+                // Move next → cur
+                if let Some(entry) = UnsentRewardEra::<T>::get(next) {
+                    UnsentRewardEra::<T>::insert(cur, entry);
+                }
+                cur = next;
+            }
+            // Remove the now-duplicate last entry and shrink tail
+            UnsentRewardEra::<T>::remove(cur);
+            let new_tail = if tail == 0 {
+                UNSENT_QUEUE_CAPACITY - 1
+            } else {
+                tail - 1
+            };
+            UnsentRewardTail::<T>::put(new_tail);
+
+            // If head was after the removed slot, adjust it too
+            let head = UnsentRewardHead::<T>::get();
+            // We also need to handle head potentially pointing past the buffer
+            // after a removal. Since we shifted everything between slot..tail back,
+            // the head only needs adjustment if it was == tail (now new_tail) — but
+            // that means the buffer just became empty, which is fine (head == new_tail).
+            // However, if head was pointing *at* a slot beyond the removed one, the
+            // entry it pointed to slid back by one, so head should also slide back.
+            // In practice, removal only happens when we know the slot, so we can
+            // simply recalculate emptiness.
+            if head == tail {
+                // Was already at tail, buffer must be empty now
+                UnsentRewardHead::<T>::put(new_tail);
+            }
+        }
+
+        // ── Core retry logic ──────────────────────────────────────────────
+
+        /// Process at most one unsent reward era per block.
+        /// On failure the head pointer advances to the next entry so a single
+        /// stuck era does not block retries for subsequent eras.
+        pub(crate) fn process_unsent_reward_eras() -> Weight {
+            let head = UnsentRewardHead::<T>::get();
+            let tail = UnsentRewardTail::<T>::get();
+
+            if head == tail {
+                return T::WeightInfo::process_unsent_reward_eras_empty();
+            }
+
+            let Some((era_index, timestamp, inflation)) = UnsentRewardEra::<T>::get(head) else {
+                // Slot unexpectedly empty — advance head past it
+                UnsentRewardHead::<T>::put((head + 1) % UNSENT_QUEUE_CAPACITY);
+                return T::WeightInfo::process_unsent_reward_eras_empty();
+            };
+
+            // Check if reward points are still available
+            let reward_points = RewardPointsForEra::<T>::get(era_index);
+            let info =
+                match reward_points.generate_era_rewards_info(era_index, inflation, timestamp) {
+                    Some(info) => info,
+                    None => {
+                        // Reward points have been pruned — discard this entry
+                        log::warn!(
+                            target: "ext_validators_rewards",
+                            "Unsent era {era_index} expired: reward points pruned",
+                        );
+                        UnsentRewardEra::<T>::remove(head);
+                        UnsentRewardHead::<T>::put((head + 1) % UNSENT_QUEUE_CAPACITY);
+                        Self::deposit_event(Event::UnsentEraExpired { era_index });
+                        return T::WeightInfo::process_unsent_reward_eras_expired();
+                    }
+                };
+
+            // Attempt to resend
+            match Self::send_rewards_message(&info) {
+                Some(message_id) => {
+                    UnsentRewardEra::<T>::remove(head);
+                    UnsentRewardHead::<T>::put((head + 1) % UNSENT_QUEUE_CAPACITY);
+                    Self::deposit_event(Event::RewardsMessageRetried {
+                        message_id,
+                        era_index,
+                        total_points: info.total_points,
+                        inflation_amount: inflation,
+                    });
+                    T::WeightInfo::process_unsent_reward_eras_success()
+                }
+                None => {
+                    // Move the failed entry to the back of the queue so the
+                    // next block tries a different era (avoids head-of-line
+                    // blocking). The entry is not lost — it will be retried
+                    // after all other pending entries.
+                    UnsentRewardEra::<T>::remove(head);
+                    UnsentRewardHead::<T>::put((head + 1) % UNSENT_QUEUE_CAPACITY);
+                    UnsentRewardEra::<T>::insert(tail, (era_index, timestamp, inflation));
+                    UnsentRewardTail::<T>::put((tail + 1) % UNSENT_QUEUE_CAPACITY);
+                    log::warn!(
+                        target: "ext_validators_rewards",
+                        "Retry for unsent era {era_index} still failing, moved to back of queue",
+                    );
+                    T::WeightInfo::process_unsent_reward_eras_failed()
+                }
+            }
+        }
+
         /// Track a block authored by a validator
         pub fn note_block_author(author: T::AccountId) {
             // Track per-session authorship for performance points
@@ -619,6 +869,24 @@ pub mod pallet {
 
             RewardPointsForEra::<T>::remove(era_index_to_delete);
             BlocksProducedInEra::<T>::remove(era_index_to_delete);
+
+            // Proactively clean up any unsent entries whose reward points
+            // have been pruned (this era and any older ones still lingering).
+            let head = UnsentRewardHead::<T>::get();
+            let mut tail = UnsentRewardTail::<T>::get();
+            let mut slot = head;
+            while slot != tail {
+                if let Some((idx, _, _)) = UnsentRewardEra::<T>::get(slot) {
+                    if idx <= era_index_to_delete {
+                        Self::unsent_queue_remove_slot(slot);
+                        tail = UnsentRewardTail::<T>::get();
+                        Self::deposit_event(Event::UnsentEraExpired { era_index: idx });
+                        // Don't advance slot — next entry slid into this position
+                        continue;
+                    }
+                }
+                slot = (slot + 1) % UNSENT_QUEUE_CAPACITY;
+            }
         }
     }
 
@@ -671,17 +939,17 @@ pub mod pallet {
 
             // Generate era rewards utils with the actual rewards amount (post-treasury split).
             // This ensures the message to EigenLayer matches the actual minted rewards.
-            let utils = match era_reward_points.generate_era_rewards_utils(
+            let info = match RewardPointsForEra::<T>::get(&era_index).generate_era_rewards_info(
                 era_index,
                 mint_result.rewards_amount,
                 era_start_timestamp,
             ) {
-                Some(utils) => utils,
+                Some(info) => info,
                 None => {
                     // Returns None when total_points is zero or no validators have rewards
                     log::error!(
                         target: "ext_validators_rewards",
-                        "Failed to generate era rewards utils (no rewards to distribute)"
+                        "Failed to generate era rewards info (no rewards to distribute)"
                     );
                     return;
                 }
@@ -692,13 +960,31 @@ pub mod pallet {
                 DispatchClass::Mandatory,
             );
 
-            if let Some(message_id) = Self::send_rewards_message(&utils) {
-                Self::deposit_event(Event::RewardsMessageSent {
-                    message_id,
-                    era_index,
-                    total_points: utils.total_points,
-                    inflation_amount: mint_result.rewards_amount,
-                });
+            match Self::send_rewards_message(&info) {
+                Some(message_id) => {
+                    Self::deposit_event(Event::RewardsMessageSent {
+                        message_id,
+                        era_index,
+                        total_points: info.total_points,
+                        inflation_amount: mint_result.rewards_amount,
+                    });
+                }
+                None => {
+                    // Message failed — queue for automatic retry via on_initialize
+                    if Self::unsent_queue_push((
+                        era_index,
+                        era_start_timestamp,
+                        mint_result.rewards_amount,
+                    )) {
+                        Self::deposit_event(Event::RewardsMessageSendFailed { era_index });
+                    } else {
+                        log::error!(
+                            target: "ext_validators_rewards",
+                            "Unsent reward queue full, cannot enqueue era {era_index}",
+                        );
+                        Self::deposit_event(Event::UnsentQueueFull { era_index });
+                    }
+                }
             }
         }
     }

operator/pallets/external-validators-rewards/src/mock.rs

@@ -131,6 +131,9 @@ impl crate::types::SendMessage for MockOkOutboundQueue {
     }
 
     fn validate(ticket: Self::Ticket) -> Result<Self::Ticket, SendError> {
+        if Mock::mock().send_message_fails {
+            return Err(SendError::MessageTooLarge);
+        }
         Ok(ticket)
     }
 
@@ -223,6 +226,7 @@ impl pallet_external_validators_rewards::Config for Test {
     type HandleInflation = InflationMinter;
     type Currency = Balances;
     type RewardsEthereumSovereignAccount = RewardsEthereumSovereignAccount;
+    type GovernanceOrigin = frame_system::EnsureRoot<H160>;
     type WeightInfo = ();
     #[cfg(feature = "runtime-benchmarks")]
     type BenchmarkHelper = ();
@@ -292,6 +296,8 @@ pub mod mock_data {
         pub offline_validators: sp_std::vec::Vec<sp_core::H160>,
         /// Set of (era_index, validator_id) pairs that are slashed
         pub slashed_validators: sp_std::vec::Vec<(u32, sp_core::H160)>,
+        /// When true, MockOkOutboundQueue::validate will return Err(SendError::MessageTooLarge)
+        pub send_message_fails: bool,
     }
 
     #[pallet::config]

operator/pallets/external-validators-rewards/src/tests.rs

@@ -16,7 +16,7 @@
 
 use {
     crate::{self as pallet_external_validators_rewards, mock::*},
-    frame_support::traits::fungible::Mutate,
+    frame_support::{assert_noop, assert_ok, traits::fungible::Mutate},
     pallet_external_validators::traits::{ActiveEraInfo, OnEraEnd, OnEraStart},
     sp_core::H160,
     sp_std::collections::btree_map::BTreeMap,
@@ -165,8 +165,8 @@ fn test_on_era_end() {
         let treasury_amount = InflationTreasuryProportion::get().mul_floor(inflation);
         let rewards_amount = inflation - treasury_amount;
         // Use 0 for era_start_timestamp in tests
-        let rewards_utils = era_rewards.generate_era_rewards_utils(1, rewards_amount, 0);
-        assert!(rewards_utils.is_some());
+        let rewards_info = era_rewards.generate_era_rewards_info(1, inflation, 0);
+        assert!(rewards_info.is_some());
         System::assert_last_event(RuntimeEvent::ExternalValidatorsRewards(
             crate::Event::RewardsMessageSent {
                 message_id: Default::default(),
@@ -207,8 +207,8 @@ fn test_on_era_end_with_zero_inflation() {
         let era_rewards = pallet_external_validators_rewards::RewardPointsForEra::<Test>::get(1);
         let inflation =
             <Test as pallet_external_validators_rewards::Config>::EraInflationProvider::get();
-        let rewards_utils = era_rewards.generate_era_rewards_utils(1, inflation, 0);
-        assert!(rewards_utils.is_some());
+        let rewards_info = era_rewards.generate_era_rewards_info(1, inflation, 0);
+        assert!(rewards_info.is_some());
         // With zero inflation, no RewardsMessageSent event should be emitted
         let events = System::events();
         assert!(
@@ -246,15 +246,15 @@ fn test_on_era_end_with_zero_points() {
         ExternalValidatorsRewards::reward_by_ids(accounts_points);
         ExternalValidatorsRewards::on_era_end(1);
 
-        // When all validators have zero points, generate_era_rewards_utils should return None
+        // When all validators have zero points, generate_era_rewards_info should return None
         // to prevent inflation from being minted with no way to distribute it
         let era_rewards = pallet_external_validators_rewards::RewardPointsForEra::<Test>::get(1);
         let inflation =
             <Test as pallet_external_validators_rewards::Config>::EraInflationProvider::get();
-        let rewards_utils = era_rewards.generate_era_rewards_utils(1, inflation, 0);
+        let rewards_info = era_rewards.generate_era_rewards_info(1, inflation, 0);
         assert!(
-            rewards_utils.is_none(),
-            "generate_era_rewards_utils should return None when total_points is zero"
+            rewards_info.is_none(),
+            "generate_era_rewards_info should return None when total_points is zero"
         );
 
         // Verify no RewardsMessageSent event was emitted
@@ -3722,3 +3722,456 @@ fn test_era_end_uses_correct_era_blocks_not_session() {
         );
     })
 }
+
+// ═══════════════════════════════════════════════════════════════════════════
+// Retry mechanism tests (ring-buffer storage)
+// ═══════════════════════════════════════════════════════════════════════════
+
+/// Helper: push an entry into the unsent ring buffer via the pallet API.
+fn push_unsent(era_index: u32, timestamp: u32, inflation: u128) {
+    assert!(
+        ExternalValidatorsRewards::unsent_queue_push((era_index, timestamp, inflation)),
+        "unsent_queue_push should succeed"
+    );
+}
+
+/// Helper: return the number of entries in the unsent ring buffer.
+fn unsent_len() -> u32 {
+    ExternalValidatorsRewards::unsent_queue_len()
+}
+
+/// Helper: check if unsent queue is empty.
+fn unsent_is_empty() -> bool {
+    ExternalValidatorsRewards::unsent_queue_is_empty()
+}
+
+#[test]
+fn send_failure_queues_era() {
+    new_test_ext().execute_with(|| {
+        run_to_block(1);
+
+        Mock::mutate(|mock| {
+            mock.active_era = Some(ActiveEraInfo {
+                index: 1,
+                start: Some(30_000),
+            });
+            mock.send_message_fails = true;
+        });
+
+        // Give validators some points
+        ExternalValidatorsRewards::reward_by_ids([(H160::from_low_u64_be(1), 100)]);
+        // Author expected blocks for 100% inflation
+        for _ in 0..600 {
+            ExternalValidatorsRewards::note_block_author(H160::from_low_u64_be(1));
+        }
+
+        ExternalValidatorsRewards::on_era_end(1);
+
+        // Verify era is queued
+        assert_eq!(unsent_len(), 1);
+
+        // Verify event
+        System::assert_has_event(RuntimeEvent::ExternalValidatorsRewards(
+            crate::Event::RewardsMessageSendFailed { era_index: 1 },
+        ));
+    })
+}
+
+#[test]
+fn on_initialize_retries_and_succeeds() {
+    new_test_ext().execute_with(|| {
+        run_to_block(1);
+
+        Mock::mutate(|mock| {
+            mock.active_era = Some(ActiveEraInfo {
+                index: 1,
+                start: Some(30_000),
+            });
+        });
+
+        // Set up reward points for era 1
+        ExternalValidatorsRewards::reward_by_ids([(H160::from_low_u64_be(1), 100)]);
+
+        // Manually populate the unsent queue
+        push_unsent(1, 30, 42);
+
+        // Sending should succeed (send_message_fails is false by default)
+        System::reset_events();
+        ExternalValidatorsRewards::process_unsent_reward_eras();
+
+        // Queue should be empty
+        assert!(unsent_is_empty());
+
+        // Verify retry event
+        System::assert_has_event(RuntimeEvent::ExternalValidatorsRewards(
+            crate::Event::RewardsMessageRetried {
+                message_id: Default::default(),
+                era_index: 1,
+                total_points: 100,
+                inflation_amount: 42,
+            },
+        ));
+    })
+}
+
+#[test]
+fn on_initialize_moves_failed_entry_to_back() {
+    new_test_ext().execute_with(|| {
+        run_to_block(1);
+
+        Mock::mutate(|mock| {
+            mock.active_era = Some(ActiveEraInfo {
+                index: 2,
+                start: Some(30_000),
+            });
+            mock.send_message_fails = true;
+        });
+
+        // Set up reward points for eras 1 and 2
+        ExternalValidatorsRewards::reward_by_ids([(H160::from_low_u64_be(1), 100)]);
+        Mock::mutate(|mock| {
+            mock.active_era = Some(ActiveEraInfo {
+                index: 1,
+                start: Some(30_000),
+            });
+        });
+        ExternalValidatorsRewards::reward_by_ids([(H160::from_low_u64_be(1), 200)]);
+
+        // Push two entries: era 1 then era 2
+        push_unsent(1, 30, 42);
+        push_unsent(2, 30, 84);
+
+        // First call: tries era 1, fails, moves era 1 to back of queue
+        ExternalValidatorsRewards::process_unsent_reward_eras();
+        // Queue length stays the same (entry moved, not removed)
+        assert_eq!(unsent_len(), 2);
+
+        // Second call: tries era 2 (NOT era 1 again), fails, moves era 2 to back
+        ExternalValidatorsRewards::process_unsent_reward_eras();
+        assert_eq!(unsent_len(), 2);
+
+        // Re-enable sending
+        Mock::mutate(|mock| mock.send_message_fails = false);
+
+        // Third call: era 1 (now at front again), succeeds
+        System::reset_events();
+        ExternalValidatorsRewards::process_unsent_reward_eras();
+        assert_eq!(unsent_len(), 1);
+        System::assert_has_event(RuntimeEvent::ExternalValidatorsRewards(
+            crate::Event::RewardsMessageRetried {
+                message_id: Default::default(),
+                era_index: 1,
+                total_points: 200,
+                inflation_amount: 42,
+            },
+        ));
+
+        // Fourth call: era 2, succeeds
+        ExternalValidatorsRewards::process_unsent_reward_eras();
+        assert!(unsent_is_empty());
+    })
+}
+
+#[test]
+fn on_initialize_removes_expired_era() {
+    new_test_ext().execute_with(|| {
+        run_to_block(1);
+
+        // Populate unsent queue with era 999 but do NOT add RewardPointsForEra for it
+        push_unsent(999, 0, 42);
+
+        System::reset_events();
+        ExternalValidatorsRewards::process_unsent_reward_eras();
+
+        // Entry should be removed
+        assert!(unsent_is_empty());
+
+        // Verify expired event
+        System::assert_has_event(RuntimeEvent::ExternalValidatorsRewards(
+            crate::Event::UnsentEraExpired { era_index: 999 },
+        ));
+    })
+}
+
+#[test]
+fn on_initialize_noop_when_queue_empty() {
+    new_test_ext().execute_with(|| {
+        run_to_block(1);
+        System::reset_events();
+
+        ExternalValidatorsRewards::process_unsent_reward_eras();
+
+        // No events should be emitted
+        let events = System::events();
+        assert!(
+            events.is_empty(),
+            "No events should be emitted when unsent queue is empty"
+        );
+    })
+}
+
+#[test]
+fn on_initialize_processes_only_head() {
+    new_test_ext().execute_with(|| {
+        run_to_block(1);
+
+        Mock::mutate(|mock| {
+            mock.active_era = Some(ActiveEraInfo {
+                index: 3,
+                start: Some(30_000),
+            });
+        });
+
+        // Set up reward points for both eras
+        ExternalValidatorsRewards::reward_by_ids([(H160::from_low_u64_be(1), 100)]);
+        Mock::mutate(|mock| {
+            mock.active_era = Some(ActiveEraInfo {
+                index: 2,
+                start: Some(30_000),
+            });
+        });
+        ExternalValidatorsRewards::reward_by_ids([(H160::from_low_u64_be(2), 200)]);
+
+        // Push two entries
+        push_unsent(3, 30, 42);
+        push_unsent(2, 20, 84);
+
+        System::reset_events();
+        ExternalValidatorsRewards::process_unsent_reward_eras();
+
+        // Only the head entry (era 3) should be processed (and removed on success)
+        assert_eq!(unsent_len(), 1);
+    })
+}
+
+#[test]
+fn retry_extrinsic_success() {
+    new_test_ext().execute_with(|| {
+        run_to_block(1);
+
+        Mock::mutate(|mock| {
+            mock.active_era = Some(ActiveEraInfo {
+                index: 1,
+                start: Some(30_000),
+            });
+        });
+
+        // Set up reward points
+        ExternalValidatorsRewards::reward_by_ids([(H160::from_low_u64_be(1), 100)]);
+
+        // Populate unsent queue
+        push_unsent(1, 30, 42);
+
+        System::reset_events();
+        assert_ok!(ExternalValidatorsRewards::retry_unsent_reward_era(
+            RuntimeOrigin::root(),
+            1
+        ));
+
+        // Queue should be empty
+        assert!(unsent_is_empty());
+
+        // Verify retry event
+        System::assert_has_event(RuntimeEvent::ExternalValidatorsRewards(
+            crate::Event::RewardsMessageRetried {
+                message_id: Default::default(),
+                era_index: 1,
+                total_points: 100,
+                inflation_amount: 42,
+            },
+        ));
+    })
+}
+
+#[test]
+fn retry_extrinsic_era_not_in_queue() {
+    new_test_ext().execute_with(|| {
+        run_to_block(1);
+
+        assert_noop!(
+            ExternalValidatorsRewards::retry_unsent_reward_era(RuntimeOrigin::root(), 1),
+            crate::Error::<Test>::EraNotInUnsentQueue
+        );
+    })
+}
+
+#[test]
+fn retry_extrinsic_pruned_data() {
+    new_test_ext().execute_with(|| {
+        run_to_block(1);
+
+        // Queue an era but don't create reward points for it
+        push_unsent(999, 0, 42);
+
+        assert_noop!(
+            ExternalValidatorsRewards::retry_unsent_reward_era(RuntimeOrigin::root(), 999),
+            crate::Error::<Test>::RewardPointsPruned
+        );
+    })
+}
+
+#[test]
+fn retry_extrinsic_requires_root() {
+    new_test_ext().execute_with(|| {
+        run_to_block(1);
+
+        assert_noop!(
+            ExternalValidatorsRewards::retry_unsent_reward_era(
+                RuntimeOrigin::signed(H160::from_low_u64_be(1)),
+                1
+            ),
+            sp_runtime::DispatchError::BadOrigin
+        );
+    })
+}
+
+#[test]
+fn unsent_queue_full() {
+    new_test_ext().execute_with(|| {
+        run_to_block(1);
+
+        Mock::mutate(|mock| {
+            mock.active_era = Some(ActiveEraInfo {
+                index: 65,
+                start: Some(30_000),
+            });
+            mock.send_message_fails = true;
+        });
+
+        // Fill the ring buffer to capacity (63 entries, since capacity=64
+        // means 63 usable slots in a ring buffer with head==tail==empty).
+        for i in 0..63u32 {
+            push_unsent(i, 0, 42);
+        }
+        assert_eq!(unsent_len(), 63);
+
+        // Give validators some points so on_era_end doesn't bail early
+        ExternalValidatorsRewards::reward_by_ids([(H160::from_low_u64_be(1), 100)]);
+        for _ in 0..600 {
+            ExternalValidatorsRewards::note_block_author(H160::from_low_u64_be(1));
+        }
+
+        System::reset_events();
+        ExternalValidatorsRewards::on_era_end(65);
+
+        // Verify UnsentQueueFull event
+        System::assert_has_event(RuntimeEvent::ExternalValidatorsRewards(
+            crate::Event::UnsentQueueFull { era_index: 65 },
+        ));
+
+        // Queue should still be at 63
+        assert_eq!(unsent_len(), 63);
+    })
+}
+
+#[test]
+fn on_era_start_prunes_unsent_entry() {
+    new_test_ext().execute_with(|| {
+        run_to_block(1);
+
+        // Set up: era 1 has an unsent entry
+        push_unsent(1, 0, 42);
+
+        // HistoryDepth is 10, so era 11 should prune era 1
+        System::reset_events();
+        ExternalValidatorsRewards::on_era_start(11, 0, 11);
+
+        // Unsent entry should be removed
+        assert!(unsent_is_empty());
+
+        // Verify expired event
+        System::assert_has_event(RuntimeEvent::ExternalValidatorsRewards(
+            crate::Event::UnsentEraExpired { era_index: 1 },
+        ));
+    })
+}
+
+#[test]
+fn retry_extrinsic_send_still_fails() {
+    new_test_ext().execute_with(|| {
+        run_to_block(1);
+
+        Mock::mutate(|mock| {
+            mock.active_era = Some(ActiveEraInfo {
+                index: 1,
+                start: Some(30_000),
+            });
+            mock.send_message_fails = true;
+        });
+
+        // Set up reward points
+        ExternalValidatorsRewards::reward_by_ids([(H160::from_low_u64_be(1), 100)]);
+
+        // Populate unsent queue
+        push_unsent(1, 30, 42);
+
+        assert_noop!(
+            ExternalValidatorsRewards::retry_unsent_reward_era(RuntimeOrigin::root(), 1),
+            crate::Error::<Test>::MessageSendFailed
+        );
+
+        // Queue should still have the entry
+        assert_eq!(unsent_len(), 1);
+    })
+}
+
+#[test]
+fn head_of_line_blocking_avoided() {
+    new_test_ext().execute_with(|| {
+        run_to_block(1);
+
+        // Set up reward points for eras 1, 2, 3
+        for era in 1..=3u32 {
+            Mock::mutate(|mock| {
+                mock.active_era = Some(ActiveEraInfo {
+                    index: era,
+                    start: Some(30_000),
+                });
+            });
+            ExternalValidatorsRewards::reward_by_ids([(H160::from_low_u64_be(1), 100)]);
+        }
+
+        // Push eras 1, 2, 3 into the queue
+        push_unsent(1, 30, 10);
+        push_unsent(2, 30, 20);
+        push_unsent(3, 30, 30);
+
+        // Make sending fail
+        Mock::mutate(|mock| mock.send_message_fails = true);
+
+        // Block 1: tries era 1, fails, advances head → era 2
+        ExternalValidatorsRewards::process_unsent_reward_eras();
+        // Block 2: tries era 2, fails, advances head → era 3
+        ExternalValidatorsRewards::process_unsent_reward_eras();
+
+        // Now re-enable sending
+        Mock::mutate(|mock| mock.send_message_fails = false);
+
+        // Block 3: tries era 3, succeeds
+        System::reset_events();
+        ExternalValidatorsRewards::process_unsent_reward_eras();
+        System::assert_has_event(RuntimeEvent::ExternalValidatorsRewards(
+            crate::Event::RewardsMessageRetried {
+                message_id: Default::default(),
+                era_index: 3,
+                total_points: 100,
+                inflation_amount: 30,
+            },
+        ));
+
+        // Block 4: wraps around to era 1, succeeds
+        ExternalValidatorsRewards::process_unsent_reward_eras();
+        System::assert_has_event(RuntimeEvent::ExternalValidatorsRewards(
+            crate::Event::RewardsMessageRetried {
+                message_id: Default::default(),
+                era_index: 1,
+                total_points: 100,
+                inflation_amount: 10,
+            },
+        ));
+
+        // Block 5: era 2, succeeds
+        ExternalValidatorsRewards::process_unsent_reward_eras();
+        assert!(unsent_is_empty());
+    })
+}

operator/pallets/external-validators-rewards/src/weights.rs

@@ -54,6 +54,11 @@ use sp_std::marker::PhantomData;
 /// Weight functions needed for pallet_external_validators_rewards.
 pub trait WeightInfo {
 	fn on_era_end() -> Weight;
+	fn process_unsent_reward_eras_empty() -> Weight;
+	fn process_unsent_reward_eras_expired() -> Weight;
+	fn process_unsent_reward_eras_success() -> Weight;
+	fn process_unsent_reward_eras_failed() -> Weight;
+	fn retry_unsent_reward_era() -> Weight;
 }
 
 /// Weights for pallet_external_validators_rewards using the Substrate node and recommended hardware.
@@ -84,6 +89,36 @@ impl<T: frame_system::Config> WeightInfo for SubstrateWeight<T> {
 			.saturating_add(T::DbWeight::get().reads(5_u64))
 			.saturating_add(T::DbWeight::get().writes(5_u64))
 	}
+
+	fn process_unsent_reward_eras_empty() -> Weight {
+		// 1 read for UnsentRewardEras
+		Weight::from_parts(5_000_000, 0)
+			.saturating_add(T::DbWeight::get().reads(1_u64))
+	}
+
+	fn process_unsent_reward_eras_expired() -> Weight {
+		// 1 read UnsentRewardEras + 1 read RewardPointsForEra + 1 write UnsentRewardEras
+		Weight::from_parts(10_000_000, 0)
+			.saturating_add(T::DbWeight::get().reads(2_u64))
+			.saturating_add(T::DbWeight::get().writes(1_u64))
+	}
+
+	fn process_unsent_reward_eras_success() -> Weight {
+		// Same as on_era_end + queue read/write
+		Weight::from_parts(1_136_401_000, 39987)
+			.saturating_add(T::DbWeight::get().reads(7_u64))
+			.saturating_add(T::DbWeight::get().writes(6_u64))
+	}
+
+	fn process_unsent_reward_eras_failed() -> Weight {
+		// Use success weight as upper bound
+		Self::process_unsent_reward_eras_success()
+	}
+
+	fn retry_unsent_reward_era() -> Weight {
+		// Same as success path
+		Self::process_unsent_reward_eras_success()
+	}
 }
 
 // For backwards compatibility and tests
@@ -113,4 +148,29 @@ impl WeightInfo for () {
 			.saturating_add(RocksDbWeight::get().reads(5_u64))
 			.saturating_add(RocksDbWeight::get().writes(5_u64))
 	}
+
+	fn process_unsent_reward_eras_empty() -> Weight {
+		Weight::from_parts(5_000_000, 0)
+			.saturating_add(RocksDbWeight::get().reads(1_u64))
+	}
+
+	fn process_unsent_reward_eras_expired() -> Weight {
+		Weight::from_parts(10_000_000, 0)
+			.saturating_add(RocksDbWeight::get().reads(2_u64))
+			.saturating_add(RocksDbWeight::get().writes(1_u64))
+	}
+
+	fn process_unsent_reward_eras_success() -> Weight {
+		Weight::from_parts(1_136_401_000, 39987)
+			.saturating_add(RocksDbWeight::get().reads(7_u64))
+			.saturating_add(RocksDbWeight::get().writes(6_u64))
+	}
+
+	fn process_unsent_reward_eras_failed() -> Weight {
+		Self::process_unsent_reward_eras_success()
+	}
+
+	fn retry_unsent_reward_era() -> Weight {
+		Self::process_unsent_reward_eras_success()
+	}
 }

operator/runtime/mainnet/src/configs/mod.rs

@@ -1598,6 +1598,8 @@ impl pallet_external_validators_rewards::Config for Runtime {
     type RewardsEthereumSovereignAccount = ExternalValidatorRewardsAccount;
     type SendMessage = RewardsSendAdapter;
     type HandleInflation = ExternalRewardsInflationHandler;
+    type GovernanceOrigin =
+        EitherOfDiverse<EnsureRoot<AccountId>, governance::custom_origins::GeneralAdmin>;
     type WeightInfo = mainnet_weights::pallet_external_validators_rewards::WeightInfo<Runtime>;
     #[cfg(feature = "runtime-benchmarks")]
     type BenchmarkHelper = ();

operator/runtime/mainnet/src/weights/pallet_external_validators_rewards.rs

@@ -74,4 +74,29 @@ impl<T: frame_system::Config> pallet_external_validators_rewards::WeightInfo for
 			.saturating_add(T::DbWeight::get().reads(9_u64))
 			.saturating_add(T::DbWeight::get().writes(2_u64))
 	}
+
+	fn process_unsent_reward_eras_empty() -> Weight {
+		Weight::from_parts(5_000_000, 0)
+			.saturating_add(T::DbWeight::get().reads(1_u64))
+	}
+
+	fn process_unsent_reward_eras_expired() -> Weight {
+		Weight::from_parts(10_000_000, 0)
+			.saturating_add(T::DbWeight::get().reads(2_u64))
+			.saturating_add(T::DbWeight::get().writes(1_u64))
+	}
+
+	fn process_unsent_reward_eras_success() -> Weight {
+		Weight::from_parts(1_905_623_000, 29162)
+			.saturating_add(T::DbWeight::get().reads(11_u64))
+			.saturating_add(T::DbWeight::get().writes(3_u64))
+	}
+
+	fn process_unsent_reward_eras_failed() -> Weight {
+		Self::process_unsent_reward_eras_success()
+	}
+
+	fn retry_unsent_reward_era() -> Weight {
+		Self::process_unsent_reward_eras_success()
+	}
 }

operator/runtime/stagenet/src/configs/mod.rs

@@ -1594,6 +1594,8 @@ impl pallet_external_validators_rewards::Config for Runtime {
     type RewardsEthereumSovereignAccount = ExternalValidatorRewardsAccount;
     type SendMessage = RewardsSendAdapter;
     type HandleInflation = ExternalRewardsInflationHandler;
+    type GovernanceOrigin =
+        EitherOfDiverse<EnsureRoot<AccountId>, governance::custom_origins::GeneralAdmin>;
     type WeightInfo = stagenet_weights::pallet_external_validators_rewards::WeightInfo<Runtime>;
     #[cfg(feature = "runtime-benchmarks")]
     type BenchmarkHelper = ();

operator/runtime/stagenet/src/weights/pallet_external_validators_rewards.rs

@@ -74,4 +74,29 @@ impl<T: frame_system::Config> pallet_external_validators_rewards::WeightInfo for
 			.saturating_add(T::DbWeight::get().reads(9_u64))
 			.saturating_add(T::DbWeight::get().writes(2_u64))
 	}
+
+	fn process_unsent_reward_eras_empty() -> Weight {
+		Weight::from_parts(5_000_000, 0)
+			.saturating_add(T::DbWeight::get().reads(1_u64))
+	}
+
+	fn process_unsent_reward_eras_expired() -> Weight {
+		Weight::from_parts(10_000_000, 0)
+			.saturating_add(T::DbWeight::get().reads(2_u64))
+			.saturating_add(T::DbWeight::get().writes(1_u64))
+	}
+
+	fn process_unsent_reward_eras_success() -> Weight {
+		Weight::from_parts(1_894_953_000, 29162)
+			.saturating_add(T::DbWeight::get().reads(11_u64))
+			.saturating_add(T::DbWeight::get().writes(3_u64))
+	}
+
+	fn process_unsent_reward_eras_failed() -> Weight {
+		Self::process_unsent_reward_eras_success()
+	}
+
+	fn retry_unsent_reward_era() -> Weight {
+		Self::process_unsent_reward_eras_success()
+	}
 }

operator/runtime/testnet/src/configs/mod.rs

@@ -1598,6 +1598,8 @@ impl pallet_external_validators_rewards::Config for Runtime {
     type RewardsEthereumSovereignAccount = ExternalValidatorRewardsAccount;
     type SendMessage = RewardsSendAdapter;
     type HandleInflation = ExternalRewardsInflationHandler;
+    type GovernanceOrigin =
+        EitherOfDiverse<EnsureRoot<AccountId>, governance::custom_origins::GeneralAdmin>;
     type WeightInfo = testnet_weights::pallet_external_validators_rewards::WeightInfo<Runtime>;
     #[cfg(feature = "runtime-benchmarks")]
     type BenchmarkHelper = ();

operator/runtime/testnet/src/weights/pallet_external_validators_rewards.rs

@@ -74,4 +74,29 @@ impl<T: frame_system::Config> pallet_external_validators_rewards::WeightInfo for
 			.saturating_add(T::DbWeight::get().reads(9_u64))
 			.saturating_add(T::DbWeight::get().writes(2_u64))
 	}
+
+	fn process_unsent_reward_eras_empty() -> Weight {
+		Weight::from_parts(5_000_000, 0)
+			.saturating_add(T::DbWeight::get().reads(1_u64))
+	}
+
+	fn process_unsent_reward_eras_expired() -> Weight {
+		Weight::from_parts(10_000_000, 0)
+			.saturating_add(T::DbWeight::get().reads(2_u64))
+			.saturating_add(T::DbWeight::get().writes(1_u64))
+	}
+
+	fn process_unsent_reward_eras_success() -> Weight {
+		Weight::from_parts(1_893_280_000, 29162)
+			.saturating_add(T::DbWeight::get().reads(11_u64))
+			.saturating_add(T::DbWeight::get().writes(3_u64))
+	}
+
+	fn process_unsent_reward_eras_failed() -> Weight {
+		Self::process_unsent_reward_eras_success()
+	}
+
+	fn retry_unsent_reward_era() -> Weight {
+		Self::process_unsent_reward_eras_success()
+	}
 }

test/.papi/descriptors/package.json

@@ -1,5 +1,5 @@
 {
-  "version": "0.1.0-autogenerated.15484599658830368838",
+  "version": "0.1.0-autogenerated.18139584469151706411",
   "name": "@polkadot-api/descriptors",
   "files": [
     "dist"

test/bun.lock

@@ -14,6 +14,9 @@
         "@noble/curves": "^1.9.2",
         "@noble/hashes": "^1.8.0",
         "@polkadot-api/descriptors": "file:.papi/descriptors",
+        "@storagehub-sdk/core": "^0.4.4",
+        "@storagehub-sdk/msp-client": "^0.4.4",
+        "@storagehub/api-augment": "^0.4.0",
         "@types/dockerode": "^3.3.41",
         "@types/node": "^22.15.32",
         "@wagmi/cli": "^2.3.1",
@@ -568,6 +571,14 @@
 
     "@sqltools/formatter": ["@sqltools/formatter@1.2.5", "", {}, "sha512-Uy0+khmZqUrUGm5dmMqVlnvufZRSK0FbYzVgp0UMstm+F5+W2/jnEEQyc9vo1ZR/E5ZI/B1WjjoTqBqwJL6Krw=="],
 
+    "@storagehub-sdk/core": ["@storagehub-sdk/core@0.4.4", "", { "dependencies": { "@polkadot/types": "^16.4.7", "abitype": "^1.0.0", "ethers": "^6.15.0" }, "peerDependencies": { "viem": ">=2.38.3" } }, "sha512-3tvsp5ILx4r1JWzqef02EKKL+u9nZIrl+/PMpj4Ode17v+mDmYI2ME3On9fZ8/+dEIAXWgqGh8/EjkYdP9PAEQ=="],
+
+    "@storagehub-sdk/msp-client": ["@storagehub-sdk/msp-client@0.4.4", "", { "peerDependencies": { "@storagehub-sdk/core": ">=0.0.5", "viem": ">=2.38.3" } }, "sha512-7TLSQAhwJ+RFxU5SbknRw37Qkhts3u2DycdZyA7aUe6e+QyD917QNnlYcM/JJLZFFiqGwy+Nrk07xhKv1zKAZg=="],
+
+    "@storagehub/api-augment": ["@storagehub/api-augment@0.4.2", "", { "dependencies": { "@polkadot/api": "^16.4.7", "@polkadot/api-base": "^16.4.7", "@polkadot/rpc-core": "^16.4.7", "@polkadot/typegen": "^16.4.7", "@polkadot/types": "^16.4.7", "@polkadot/types-codec": "^16.4.7", "@storagehub/types-bundle": "0.4.2", "tsx": "4.20.5", "typescript": "^5.9.2" } }, "sha512-L3q5ZsZD+iLPEdBs2ZTKeH5fDaihiUJQpyxSC3pj0geOdE97m+FqxgOALEvAZT7Eqi0m38B0xneREzwPpIGtnA=="],
+
+    "@storagehub/types-bundle": ["@storagehub/types-bundle@0.4.2", "", { "dependencies": { "@polkadot/api": "^16.4.7", "@polkadot/api-base": "^16.4.7", "@polkadot/rpc-core": "^16.4.6", "@polkadot/typegen": "^16.4.6", "@polkadot/types": "^16.4.7", "@polkadot/types-codec": "^16.4.7", "typescript": "^5.9.2" } }, "sha512-kkWYP1WwiVP0NGQqIWLfcOsIkb1BJXk7Qw+pkNIzf7QW6HpJaPySJybRksK6ClwKdqzNXXyZ4Sw0vBO1//8h0w=="],
+
     "@substrate/connect": ["@substrate/connect@0.8.11", "", { "dependencies": { "@substrate/connect-extension-protocol": "^2.0.0", "@substrate/connect-known-chains": "^1.1.5", "@substrate/light-client-extension-helpers": "^1.0.0", "smoldot": "2.0.26" } }, "sha512-ofLs1PAO9AtDdPbdyTYj217Pe+lBfTLltdHDs3ds8no0BseoLeAGxpz1mHfi7zB4IxI3YyAiLjH6U8cw4pj4Nw=="],
 
     "@substrate/connect-extension-protocol": ["@substrate/connect-extension-protocol@2.2.2", "", {}, "sha512-t66jwrXA0s5Goq82ZtjagLNd7DPGCNjHeehRlE/gcJmJ+G56C0W+2plqOMRicJ8XGR1/YFnUSEqUFiSNbjGrAA=="],
@@ -2180,6 +2191,10 @@
 
     "@safe-global/safe-apps-sdk/viem": ["viem@2.29.2", "", { "dependencies": { "@noble/curves": "1.8.2", "@noble/hashes": "1.7.2", "@scure/bip32": "1.6.2", "@scure/bip39": "1.5.4", "abitype": "1.0.8", "isows": "1.0.6", "ox": "0.6.9", "ws": "8.18.1" }, "peerDependencies": { "typescript": ">=5.0.4" }, "optionalPeers": ["typescript"] }, "sha512-cukRxab90jvQ+TDD84sU3qB3UmejYqgCw4cX8SfWzvh7JPfZXI3kAMUaT5OSR2As1Mgvx1EJawccwPjGqkSSwA=="],
 
+    "@storagehub/api-augment/typescript": ["typescript@5.9.3", "", { "bin": { "tsc": "bin/tsc", "tsserver": "bin/tsserver" } }, "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw=="],
+
+    "@storagehub/types-bundle/typescript": ["typescript@5.9.3", "", { "bin": { "tsc": "bin/tsc", "tsserver": "bin/tsserver" } }, "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw=="],
+
     "@substrate/connect/smoldot": ["smoldot@2.0.26", "", { "dependencies": { "ws": "^8.8.1" } }, "sha512-F+qYmH4z2s2FK+CxGj8moYcd1ekSIKH8ywkdqlOz88Dat35iB1DIYL11aILN46YSGMzQW/lbJNS307zBSDN5Ig=="],
 
     "@substrate/light-client-extension-helpers/@polkadot-api/json-rpc-provider": ["@polkadot-api/json-rpc-provider@0.0.1", "", {}, "sha512-/SMC/l7foRjpykLTUTacIH05H3mr9ip8b5xxfwXlVezXrNVLp3Cv0GX6uItkKd+ZjzVPf3PFrDF2B2/HLSNESA=="],

test/e2e/framework/validators.ts

@@ -54,6 +54,8 @@ export const launchDatahavenValidator = async (
   const COMMON_LAUNCH_ARGS = [
     "--unsafe-force-node-key-generation",
     "--tmp",
+    "--chain",
+    "local",
     "--validator",
     "--discover-local",
     "--no-prometheus",

test/e2e/suites/storagehub.test.ts

@@ -0,0 +1,228 @@
+/**
+ * StorageHub E2E Tests
+ *
+ * Tests the uploading a file to storage through Datahaven
+ *
+ * Prerequisites:
+ * - DataHaven network with StorageHub service running
+ * - Storage hub MSP and BSP
+ */
+import "@storagehub/api-augment";
+import { afterAll, beforeAll, describe, expect, it } from "bun:test";
+import { TypeRegistry } from "@polkadot/types";
+import {
+  FileManager,
+  initWasm,
+  ReplicationLevel,
+  SH_FILE_SYSTEM_PRECOMPILE_ADDRESS,
+  StorageHubClient
+} from "@storagehub-sdk/core";
+import { MspClient } from "@storagehub-sdk/msp-client";
+import { $ } from "bun";
+import { Binary } from "polkadot-api";
+import { createPapiConnectors, logger } from "utils";
+import { CHAIN_ID, SUBSTRATE_FUNDED_ACCOUNTS } from "utils/constants";
+import { getEvmEcdsaSigner } from "utils/papi";
+import { createPublicClient, createWalletClient, defineChain, http } from "viem";
+import { privateKeyToAccount } from "viem/accounts";
+import { launchLocalDataHavenSolochain } from "../../launcher/datahaven";
+import {
+  launchBackend,
+  launchBspNode,
+  launchIndexerNode,
+  launchMspNode,
+  launchStorageHubPostgres
+} from "../../launcher/storagehub-docker";
+import { LaunchedNetwork } from "../../launcher/types/launchedNetwork";
+import { registerProviders } from "../../scripts/register-providers";
+
+const TEST_AUTHORITY_IDS = ["alice", "bob"] as const;
+const networkId = `storagehub-${Date.now()}`.toLowerCase().replace(/[^a-z0-9-]/g, "-");
+
+describe("test uploading file to storage hub", () => {
+  let aliceUrl: string;
+  let _mspUrl: string;
+  let backendUrl: string;
+
+  beforeAll(async () => {
+    await initWasm();
+
+    const datahavenImageTag = "datahavenxyz/datahaven:local";
+    const relayerImageTag = "datahavenxyz/snowbridge-relay:latest";
+    const authorityIds = TEST_AUTHORITY_IDS;
+    const buildDatahaven = false;
+    const datahavenBuildExtraArgs = "";
+
+    const options = {
+      networkId,
+      datahavenImageTag,
+      relayerImageTag,
+      authorityIds,
+      buildDatahaven,
+      datahavenBuildExtraArgs
+    };
+
+    const run = new LaunchedNetwork();
+
+    // 1. Launch DataHaven validator nodes
+    logger.info("📦 Launching DataHaven validator nodes...");
+    aliceUrl = await launchLocalDataHavenSolochain(options, run);
+
+    // 2. Launch PostgreSQL database
+    logger.info("🗄️ Launching StorageHub PostgreSQL...");
+    await launchStorageHubPostgres(options, run);
+
+    // 3. Launch MSP node
+    logger.info("📦 Launching MSP node...");
+    _mspUrl = await launchMspNode(options, run);
+
+    // 4. Launch BSP node
+    logger.info("📦 Launching BSP node...");
+    await launchBspNode(options, run);
+
+    // 6. Launch Indexer node
+    logger.info("📦 Launching Indexer node...");
+    await launchIndexerNode(options, run);
+
+    // // 7. Launch Fisherman node
+    // logger.info("📦 Launching Fisherman node...");
+    // await launchFishermanNode(options, run);
+
+    // Register providers
+    logger.info("📝 Registering providers...");
+    await registerProviders({ launchedNetwork: run });
+
+    // Launch Storage Hub Backend
+    logger.info("📦 Launching Storage hub backend...");
+    backendUrl = await launchBackend(options, run);
+  });
+
+  it("Create a bucket", async () => {
+    const { typedApi: dhApi } = createPapiConnectors(aliceUrl);
+
+    const mspCount = await dhApi.query.Providers.MspCount.getValue();
+    const bspCount = await dhApi.query.Providers.BspCount.getValue();
+
+    expect(mspCount).toBe(1);
+    expect(bspCount).toBe(1);
+
+    const msp_id = await dhApi.query.Providers.AccountIdToMainStorageProviderId.getValue(
+      SUBSTRATE_FUNDED_ACCOUNTS.CHARLETH.publicKey
+    );
+    expect(msp_id).toBeDefined();
+    if (!msp_id) {
+      throw new Error("mspId for Charleth not found");
+    }
+
+    const value_prop_id =
+      await dhApi.apis.StorageProvidersApi.query_value_propositions_for_msp(msp_id);
+
+    const call = await dhApi.tx.FileSystem.create_bucket({
+      msp_id,
+      name: Binary.fromText("bucket"),
+      private: false,
+      value_prop_id: value_prop_id[0].id
+    });
+    const aliceSigner = getEvmEcdsaSigner(SUBSTRATE_FUNDED_ACCOUNTS.ALITH.privateKey);
+    const mspResult = await call.signAndSubmit(aliceSigner);
+    expect(mspResult.ok).toBeTrue();
+  }, 30000);
+
+  it("Send a request", async () => {
+    const { typedApi: dhApi } = createPapiConnectors(aliceUrl);
+
+    const msp_id = await dhApi.query.Providers.AccountIdToMainStorageProviderId.getValue(
+      SUBSTRATE_FUNDED_ACCOUNTS.CHARLETH.publicKey
+    );
+    expect(msp_id).toBeDefined();
+    if (!msp_id) {
+      throw new Error("mspId for Charleth not found");
+    }
+
+    const buckets = await dhApi.apis.StorageProvidersApi.query_buckets_for_msp(msp_id);
+    if (!buckets.success) {
+      throw new Error("Bucket not found for the registered msp");
+    }
+    expect(buckets.value.length).toBe(1);
+
+    const bucketId = buckets.value[0].asHex();
+    const fileContent = "foo bar";
+    const location = "foo/bar.txt";
+
+    // Build FileManager from in-memory file content
+    const fileBytes = new TextEncoder().encode(fileContent);
+    const fileManager = new FileManager({
+      size: fileBytes.length,
+      stream: () =>
+        new ReadableStream({
+          start(controller) {
+            controller.enqueue(fileBytes);
+            controller.close();
+          }
+        }) as ReadableStream<Uint8Array>
+    });
+
+    // Compute fingerprint and file key from the file metadata
+    const registry = new TypeRegistry();
+    const account = privateKeyToAccount(SUBSTRATE_FUNDED_ACCOUNTS.ALITH.privateKey);
+    const owner = registry.createType("AccountId20", account.address);
+    const bucketIdH256 = registry.createType("H256", bucketId);
+    const fingerprint = await fileManager.getFingerprint();
+    const _fileKey = await fileManager.computeFileKey(owner, bucketIdH256, location);
+
+    // Set up EVM clients
+    const httpUrl = aliceUrl.replace("ws://", "http://");
+    const chain = defineChain({
+      id: CHAIN_ID,
+      name: "DataHaven",
+      nativeCurrency: { decimals: 18, name: "Ether", symbol: "ETH" },
+      rpcUrls: { default: { http: [httpUrl] } }
+    });
+    const walletClient = createWalletClient({ account, chain, transport: http(httpUrl) });
+    const publicClient = createPublicClient({ chain, transport: http(httpUrl) });
+    const storageHubClient = new StorageHubClient({
+      rpcUrl: httpUrl,
+      chain,
+      walletClient,
+      filesystemContractAddress: SH_FILE_SYSTEM_PRECOMPILE_ADDRESS
+    });
+
+    // Issue storage request
+    const txHash = await storageHubClient.issueStorageRequest(
+      bucketId as `0x${string}`,
+      location,
+      fingerprint.toHex() as `0x${string}`,
+      BigInt(fileBytes.length),
+      msp_id.asHex() as `0x${string}`,
+      [],
+      ReplicationLevel.Basic,
+      1
+    );
+
+    // Wait for storage request transaction
+    // Don't proceed until receipt is confirmed on chain
+    if (!txHash) {
+      throw new Error("Storage request transaction was not submitted");
+    }
+    const receipt = await publicClient.waitForTransactionReceipt({ hash: txHash });
+    if (receipt.status !== "success") {
+      throw new Error(`Storage request failed: ${txHash}`);
+    }
+    console.log("issueStorageRequest() txReceipt:", receipt);
+
+    // Authenticate with the backend via SIWE and upload the file
+    let sessionRef: { token: string; user: { address: string } } | undefined;
+    const sessionProvider = async () => sessionRef;
+    const mspClient = await MspClient.connect({ baseUrl: backendUrl }, sessionProvider);
+
+    const domain = new URL(backendUrl).host;
+    const siweSession = await mspClient.auth.SIWE(walletClient, domain, backendUrl);
+    const sessionToken = (siweSession as { token: string }).token;
+    expect(sessionToken).toBeDefined();
+  }, 60000);
+
+  afterAll(async () => {
+    // Delete all the containers started by this test suite
+    await $`docker container rm -f $(docker container ls -q --filter name=${networkId})`;
+  });
+});

test/launcher/datahaven.ts

@@ -84,7 +84,7 @@ export const getPortMappingForNode = (nodeId: string, networkId: string): string
 export const launchLocalDataHavenSolochain = async (
   options: DataHavenOptions,
   launchedNetwork: LaunchedNetwork
-): Promise<void> => {
+): Promise<string> => {
   logger.info("🚀 Launching DataHaven network...");
 
   invariant(options.datahavenImageTag, "❌ DataHaven image tag not defined");
@@ -165,6 +165,8 @@ export const launchLocalDataHavenSolochain = async (
   await setupDataHavenValidatorConfig(launchedNetwork, "datahaven-");
 
   logger.success(`DataHaven network started, primary node accessible on port ${alicePort}`);
+
+  return `ws://127.0.0.1:${alicePort}`;
 };
 
 /**

test/launcher/storagehub-docker.ts

@@ -124,7 +124,7 @@ export const injectStorageHubKey = async (
   // Use Bun's $ directly with docker exec (no sh -c wrapper needed)
   // This properly handles the spaces in the seed phrase
   try {
-    await $`docker exec ${containerName} datahaven-node key insert --base-path /data --key-type bcsv --scheme ecdsa --suri ${secretKey}`;
+    await $`docker exec ${containerName} datahaven-node key insert --chain local --key-type bcsv --scheme ecdsa --suri ${secretKey}`;
     logger.success("Key injected successfully");
   } catch (error) {
     logger.error(`Failed to inject key : ${error}`);
@@ -141,7 +141,7 @@ export const injectStorageHubKey = async (
 export const launchMspNode = async (
   options: DataHavenOptions,
   launchedNetwork: LaunchedNetwork
-): Promise<void> => {
+): Promise<string> => {
   logger.info("🚀 Launching StorageHub MSP node...");
 
   const containerName = `storagehub-msp-${options.networkId}`;
@@ -182,7 +182,10 @@ export const launchMspNode = async (
     "--max-storage-capacity",
     "10737418240", // 10 GiB
     "--jump-capacity",
-    "1073741824" // 1 GiB
+    "1073741824", // 1 GiB
+    "--trusted-file-transfer-server",
+    "--trusted-file-transfer-server-host",
+    "0.0.0.0" // Listen on all interfaces so the backend container can reach it
   ];
 
   logger.debug(`Executing: ${command.join(" ")}`);
@@ -217,6 +220,8 @@ export const launchMspNode = async (
   launchedNetwork.addContainer(containerName, { ws: wsPort }, { ws: DEFAULT_SUBSTRATE_WS_PORT });
 
   logger.success(`MSP node started on port ${wsPort}`);
+
+  return `ws://127.0.0.1:${wsPort}`;
 };
 
 /**
@@ -457,11 +462,12 @@ export const launchFishermanNode = async (
  *
  * @param options - Configuration options for launching the network
  * @param launchedNetwork - The launched network instance to track the node
+ * @returns The HTTP URL of the backend API (e.g. "http://127.0.0.1:8080")
  */
 export const launchBackend = async (
   options: DataHavenOptions,
   launchedNetwork: LaunchedNetwork
-): Promise<void> => {
+): Promise<string> => {
   logger.info("🚀 Launching StorageHub Backend...");
 
   const backendImage = "moonsonglabs/storage-hub-msp-backend:latest";
@@ -484,8 +490,10 @@ export const launchBackend = async (
     "-e",
     "RUST_LOG=info",
     backendImage,
-    "--chain",
-    "local",
+    "--host",
+    "0.0.0.0",
+    "--port",
+    "8080",
     "--log-format",
     "text",
     "--database-url",
@@ -507,6 +515,8 @@ export const launchBackend = async (
   launchedNetwork.addContainer(containerName, { http: apiPort }, { http: apiPort });
 
   logger.success(`StorageHub Backend container started on port ${apiPort}`);
+
+  return `http://127.0.0.1:${apiPort}`;
 };
 
 /**

test/package.json

@@ -54,6 +54,9 @@
     "@noble/curves": "^1.9.2",
     "@noble/hashes": "^1.8.0",
     "@polkadot-api/descriptors": "file:.papi/descriptors",
+    "@storagehub-sdk/core": "^0.4.4",
+    "@storagehub-sdk/msp-client": "^0.4.4",
+    "@storagehub/api-augment": "^0.4.0",
     "@types/dockerode": "^3.3.41",
     "@types/node": "^22.15.32",
     "@wagmi/cli": "^2.3.1",

test/scripts/register-providers.ts

@@ -212,7 +212,7 @@ export async function verifyProvidersRegistered(
 ): Promise<boolean> {
   logger.info("🔍 Verifying provider registration...");
 
-  const aliceContainerName = `datahaven - alice - ${options.launchedNetwork.networkId} `;
+  const aliceContainerName = `datahaven-alice-${options.launchedNetwork.networkId} `;
   const alicePort = options.launchedNetwork.getContainerPort(aliceContainerName);
 
   const { client, typedApi } = createPapiConnectors(`ws://127.0.0.1:${alicePort}`);

test/utils/docker.ts

@@ -1,5 +1,6 @@
 import { existsSync } from "node:fs";
 import { type Duplex, PassThrough, Transform } from "node:stream";
+import { $ } from "bun";
 import Docker from "dockerode";
 import invariant from "tiny-invariant";
 import { logger } from "./logger";
@@ -238,6 +239,9 @@ export const waitForContainerToStart = async (
   logger.debug(`Waiting for container ${containerName} to start...`);
   const seconds = options?.timeoutSeconds ?? 30;
 
+  // sleep 2 seconds to see if the started container didn't exit right away
+  await Bun.sleep(2000);
+
   for (let i = 0; i < seconds; i++) {
     const containers = await docker.listContainers();
     const container = containers.find((container) =>
@@ -245,10 +249,17 @@ export const waitForContainerToStart = async (
     );
     if (container) {
       logger.debug(`Container ${containerName} started after ${i} seconds`);
+      const result = await $`docker logs ${containerName}`.nothrow().quiet().text();
+      console.log(result);
+
       return;
     }
     await Bun.sleep(1000);
   }
+
+  const result = await $`docker logs ${containerName}`;
+  console.log(result);
+
   invariant(
     false,
     `❌ container ${containerName} cannot be found  in running container list after ${seconds} seconds`

test/utils/papi.ts

@@ -43,6 +43,7 @@ export const createPapiConnectors = (
 ): { client: PolkadotClient; typedApi: DataHavenApi } => {
   const url = wsUrl ?? "ws://127.0.0.1:9944";
   const client = createClient(withPolkadotSdkCompat(getWsProvider(url)));
+
   return { client, typedApi: client.getTypedApi(datahaven) };
 };
 

test/utils/validators.ts

@@ -8,6 +8,8 @@
 export const COMMON_LAUNCH_ARGS = [
   "--unsafe-force-node-key-generation",
   "--tmp",
+  "--chain",
+  "local",
   "--validator",
   "--discover-local",
   "--no-prometheus",