Ensure desired pruning behavior is held within the node

Resolves https://github.com/serai-dex/serai/issues/694.

Cargo.lock

@@ -92,9 +92,9 @@ checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923"
 
 [[package]]
 name = "alloy-chains"
-version = "0.2.18"
+version = "0.2.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bfaa9ea039a6f9304b4a593d780b1f23e1ae183acdee938b11b38795acacc9f1"
+checksum = "3ef6e7627b842406f449f83ae1a437a01cd244bc246d66f102cee9c0435ce10d"
 dependencies = [
  "alloy-primitives",
  "num_enum",
@@ -979,17 +979,6 @@ dependencies = [
  "windows-sys 0.61.2",
 ]
 
-[[package]]
-name = "async-lock"
-version = "3.4.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5fd03604047cee9b6ce9de9f70c6cd540a0520c813cbd49bae61f33ab80ed1dc"
-dependencies = [
- "event-listener",
- "event-listener-strategy",
- "pin-project-lite",
-]
-
 [[package]]
 name = "async-stream"
 version = "0.3.6"
@@ -1636,9 +1625,9 @@ dependencies = [
 
 [[package]]
 name = "clap"
-version = "4.5.51"
+version = "4.5.52"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4c26d721170e0295f191a69bd9a1f93efcdb0aff38684b61ab5750468972e5f5"
+checksum = "aa8120877db0e5c011242f96806ce3c94e0737ab8108532a76a3300a01db2ab8"
 dependencies = [
  "clap_builder",
  "clap_derive",
@@ -1646,9 +1635,9 @@ dependencies = [
 
 [[package]]
 name = "clap_builder"
-version = "4.5.51"
+version = "4.5.52"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "75835f0c7bf681bfd05abe44e965760fea999a5286c6eb2d59883634fd02011a"
+checksum = "02576b399397b659c26064fbc92a75fede9d18ffd5f80ca1cd74ddab167016e1"
 dependencies = [
  "anstream",
  "anstyle",
@@ -6750,9 +6739,9 @@ checksum = "7a2d987857b319362043e95f5353c0535c1f58eec5336fdfcf626430af7def58"
 
 [[package]]
 name = "resolv-conf"
-version = "0.7.5"
+version = "0.7.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6b3789b30bd25ba102de4beabd95d21ac45b69b1be7d14522bab988c526d6799"
+checksum = "1e061d1b48cb8d38042de4ae0a7a6401009d6143dc80d2e2d6f31f0bdd6470c7"
 
 [[package]]
 name = "revm"
@@ -8334,7 +8323,6 @@ dependencies = [
 name = "serai-client-serai"
 version = "0.1.0"
 dependencies = [
- "async-lock",
  "bitvec",
  "blake2 0.11.0-rc.3",
  "borsh",
@@ -8451,7 +8439,6 @@ dependencies = [
  "serai-db",
  "serai-processor-messages",
  "serai-task",
- "tokio",
 ]
 
 [[package]]

coordinator/cosign/src/intend.rs

@@ -4,11 +4,18 @@ use std::{sync::Arc, collections::HashMap};
 use blake2::{Digest, Blake2b256};
 
 use serai_client_serai::{
-  abi::primitives::{
+  abi::{
-    balance::Amount, validator_sets::ExternalValidatorSet, address::SeraiAddress,
+    primitives::{
+      network_id::{ExternalNetworkId, NetworkId},
+      balance::Amount,
+      crypto::Public,
+      validator_sets::{Session, ExternalValidatorSet},
+      address::SeraiAddress,
       merkle::IncrementalUnbalancedMerkleTree,
     },
-  Serai,
+    validator_sets::Event,
+  },
+  Serai, Events,
 };
 
 use serai_db::*;
@@ -16,10 +23,20 @@ use serai_task::ContinuallyRan;
 
 use crate::*;
 
+#[derive(BorshSerialize, BorshDeserialize)]
+struct Set {
+  session: Session,
+  key: Public,
+  stake: Amount,
+}
+
 create_db!(
   CosignIntend {
     ScanCosignFrom: () -> u64,
     BuildsUpon: () -> IncrementalUnbalancedMerkleTree,
+    Stakes: (network: ExternalNetworkId, validator: SeraiAddress) -> Amount,
+    Validators: (set: ExternalValidatorSet) -> Vec<SeraiAddress>,
+    LatestSet: (network: ExternalNetworkId) -> Set,
   }
 );
 
@@ -40,23 +57,38 @@ db_channel! {
 async fn block_has_events_justifying_a_cosign(
   serai: &Serai,
   block_number: u64,
-) -> Result<(Block, HasEvents), String> {
+) -> Result<(Block, Events, HasEvents), String> {
   let block = serai
     .block_by_number(block_number)
     .await
     .map_err(|e| format!("{e:?}"))?
     .ok_or_else(|| "couldn't get block which should've been finalized".to_string())?;
-  let serai = serai.as_of(block.header.hash()).await.map_err(|e| format!("{e:?}"))?;
+  let events = serai.events(block.header.hash()).await.map_err(|e| format!("{e:?}"))?;
 
-  if !serai.validator_sets().set_keys_events().await.map_err(|e| format!("{e:?}"))?.is_empty() {
+  if events.validator_sets().set_keys_events().next().is_some() {
-    return Ok((block, HasEvents::Notable));
+    return Ok((block, events, HasEvents::Notable));
   }
 
-  if !serai.coins().burn_with_instruction_events().await.map_err(|e| format!("{e:?}"))?.is_empty() {
+  if events.coins().burn_with_instruction_events().next().is_some() {
-    return Ok((block, HasEvents::NonNotable));
+    return Ok((block, events, HasEvents::NonNotable));
   }
 
-  Ok((block, HasEvents::No))
+  Ok((block, events, HasEvents::No))
+}
+
+// Fetch the `ExternalValidatorSet`s, and their associated keys, used for cosigning as of this
+// block.
+fn cosigning_sets(getter: &impl Get) -> Vec<(ExternalValidatorSet, Public, Amount)> {
+  let mut sets = vec![];
+  for network in ExternalNetworkId::all() {
+    let Some(Set { session, key, stake }) = LatestSet::get(getter, network) else {
+      // If this network doesn't have usable keys, move on
+      continue;
+    };
+
+    sets.push((ExternalValidatorSet { network, session }, key, stake));
+  }
+  sets
 }
 
 /// A task to determine which blocks we should intend to cosign.
@@ -77,7 +109,7 @@ impl<D: Db> ContinuallyRan for CosignIntendTask<D> {
       for block_number in start_block_number ..= latest_block_number {
         let mut txn = self.db.txn();
 
-        let (block, mut has_events) =
+        let (block, events, mut has_events) =
           block_has_events_justifying_a_cosign(&self.serai, block_number)
             .await
             .map_err(|e| format!("{e:?}"))?;
@@ -105,32 +137,75 @@ impl<D: Db> ContinuallyRan for CosignIntendTask<D> {
         );
         BuildsUpon::set(&mut txn, &builds_upon);
 
+        // Update the stakes
+        for event in events.validator_sets().allocation_events() {
+          let Event::Allocation { validator, network, amount } = event else {
+            panic!("event from `allocation_events` wasn't `Event::Allocation`")
+          };
+          let Ok(network) = ExternalNetworkId::try_from(*network) else { continue };
+          let existing = Stakes::get(&txn, network, *validator).unwrap_or(Amount(0));
+          Stakes::set(&mut txn, network, *validator, &Amount(existing.0 + amount.0));
+        }
+        for event in events.validator_sets().deallocation_events() {
+          let Event::Deallocation { validator, network, amount, timeline: _ } = event else {
+            panic!("event from `deallocation_events` wasn't `Event::Deallocation`")
+          };
+          let Ok(network) = ExternalNetworkId::try_from(*network) else { continue };
+          let existing = Stakes::get(&txn, network, *validator).unwrap_or(Amount(0));
+          Stakes::set(&mut txn, network, *validator, &Amount(existing.0 - amount.0));
+        }
+
+        // Handle decided sets
+        for event in events.validator_sets().set_decided_events() {
+          let Event::SetDecided { set, validators } = event else {
+            panic!("event from `set_decided_events` wasn't `Event::SetDecided`")
+          };
+
+          let Ok(set) = ExternalValidatorSet::try_from(*set) else { continue };
+          Validators::set(
+            &mut txn,
+            set,
+            &validators.iter().map(|(validator, _key_shares)| *validator).collect(),
+          );
+        }
+
+        // Handle declarations of the latest set
+        for event in events.validator_sets().set_keys_events() {
+          let Event::SetKeys { set, key_pair } = event else {
+            panic!("event from `set_keys_events` wasn't `Event::SetKeys`")
+          };
+          let mut stake = 0;
+          for validator in
+            Validators::take(&mut txn, *set).expect("set which wasn't decided set keys")
+          {
+            stake += Stakes::get(&txn, set.network, validator).unwrap_or(Amount(0)).0;
+          }
+          LatestSet::set(
+            &mut txn,
+            set.network,
+            &Set { session: set.session, key: key_pair.0, stake: Amount(stake) },
+          );
+        }
+
         let global_session_for_this_block = LatestGlobalSessionIntended::get(&txn);
 
         // If this is notable, it creates a new global session, which we index into the database
         // now
         if has_events == HasEvents::Notable {
-          let serai = self.serai.as_of(block_hash).await.map_err(|e| format!("{e:?}"))?;
+          let sets_and_keys_and_stakes = cosigning_sets(&txn);
-          let sets_and_keys = cosigning_sets(&serai).await?;
+          let global_session = GlobalSession::id(
-          let global_session =
+            sets_and_keys_and_stakes.iter().map(|(set, _key, _stake)| *set).collect(),
-            GlobalSession::id(sets_and_keys.iter().map(|(set, _key)| *set).collect());
+          );
 
-          let mut sets = Vec::with_capacity(sets_and_keys.len());
+          let mut sets = Vec::with_capacity(sets_and_keys_and_stakes.len());
-          let mut keys = HashMap::with_capacity(sets_and_keys.len());
+          let mut keys = HashMap::with_capacity(sets_and_keys_and_stakes.len());
-          let mut stakes = HashMap::with_capacity(sets_and_keys.len());
+          let mut stakes = HashMap::with_capacity(sets_and_keys_and_stakes.len());
           let mut total_stake = 0;
-          for (set, key) in &sets_and_keys {
+          for (set, key, stake) in sets_and_keys_and_stakes {
-            sets.push(*set);
+            sets.push(set);
-            keys.insert(set.network, SeraiAddress::from(*key));
+            keys.insert(set.network, key);
-            let stake = serai
+            stakes.insert(set.network, stake.0);
-              .validator_sets()
+            total_stake += stake.0;
-              .current_stake(set.network.into())
-              .await
-              .map_err(|e| format!("{e:?}"))?
-              .unwrap_or(Amount(0))
-              .0;
-            stakes.insert(set.network, stake);
-            total_stake += stake;
           }
           if total_stake == 0 {
             Err(format!("cosigning sets for block #{block_number} had 0 stake in total"))?;

coordinator/cosign/src/lib.rs

@@ -20,7 +20,7 @@ use serai_client_serai::{
     },
     Block,
   },
-  Serai, TemporalSerai,
+  Serai, State,
 };
 
 use serai_db::*;
@@ -59,7 +59,7 @@ use delay::LatestCosignedBlockNumber;
 pub(crate) struct GlobalSession {
   pub(crate) start_block_number: u64,
   pub(crate) sets: Vec<ExternalValidatorSet>,
-  pub(crate) keys: HashMap<ExternalNetworkId, SeraiAddress>,
+  pub(crate) keys: HashMap<ExternalNetworkId, Public>,
   pub(crate) stakes: HashMap<ExternalNetworkId, u64>,
   pub(crate) total_stake: u64,
 }
@@ -121,60 +121,6 @@ create_db! {
   }
 }
 
-/// Fetch the keys used for cosigning by a specific network.
-async fn keys_for_network(
-  serai: &TemporalSerai<'_>,
-  network: ExternalNetworkId,
-) -> Result<Option<(Session, KeyPair)>, String> {
-  let Some(latest_session) =
-    serai.validator_sets().current_session(network.into()).await.map_err(|e| format!("{e:?}"))?
-  else {
-    // If this network hasn't had a session declared, move on
-    return Ok(None);
-  };
-
-  // Get the keys for the latest session
-  if let Some(keys) = serai
-    .validator_sets()
-    .keys(ExternalValidatorSet { network, session: latest_session })
-    .await
-    .map_err(|e| format!("{e:?}"))?
-  {
-    return Ok(Some((latest_session, keys)));
-  }
-
-  // If the latest session has yet to set keys, use the prior session
-  if let Some(prior_session) = latest_session.0.checked_sub(1).map(Session) {
-    if let Some(keys) = serai
-      .validator_sets()
-      .keys(ExternalValidatorSet { network, session: prior_session })
-      .await
-      .map_err(|e| format!("{e:?}"))?
-    {
-      return Ok(Some((prior_session, keys)));
-    }
-  }
-
-  Ok(None)
-}
-
-/// Fetch the `ExternalValidatorSet`s, and their associated keys, used for cosigning as of this
-/// block.
-async fn cosigning_sets(
-  serai: &TemporalSerai<'_>,
-) -> Result<Vec<(ExternalValidatorSet, Public)>, String> {
-  let mut sets = vec![];
-  for network in ExternalNetworkId::all() {
-    let Some((session, keys)) = keys_for_network(serai, network).await? else {
-      // If this network doesn't have usable keys, move on
-      continue;
-    };
-
-    sets.push((ExternalValidatorSet { network, session }, keys.0));
-  }
-  Ok(sets)
-}
-
 /// An object usable to request notable cosigns for a block.
 pub trait RequestNotableCosigns: 'static + Send {
   /// The error type which may be encountered when requesting notable cosigns.
@@ -379,13 +325,8 @@ impl<D: Db> Cosigning<D> {
 
     // Check the cosign's signature
     {
-      let key = Public::from({
+      let key =
-        let Some(key) = global_session.keys.get(&network) else {
+        *global_session.keys.get(&network).ok_or(IntakeCosignError::NonParticipatingNetwork)?;
-          Err(IntakeCosignError::NonParticipatingNetwork)?
-        };
-        *key
-      });
-
       if !signed_cosign.verify_signature(key) {
         Err(IntakeCosignError::InvalidSignature)?;
       }

coordinator/p2p/libp2p/src/validators.rs

@@ -60,8 +60,7 @@ impl Validators {
 
       Besides, we can't connect to historical validators, only the current validators.
     */
-    let temporal_serai = serai.borrow().as_of_latest_finalized_block().await?;
+    let serai = serai.borrow().state().await?;
-    let temporal_serai = temporal_serai.validator_sets();
 
     let mut session_changes = vec![];
     {
@@ -70,9 +69,9 @@ impl Validators {
       let mut futures = FuturesUnordered::new();
       for network in ExternalNetworkId::all() {
         let sessions = sessions.borrow();
-        let temporal_serai = temporal_serai.borrow();
+        let serai = serai.borrow();
         futures.push(async move {
-          let session = match temporal_serai.current_session(network.into()).await {
+          let session = match serai.current_session(network.into()).await {
             Ok(Some(session)) => session,
             Ok(None) => return Ok(None),
             Err(e) => return Err(e),
@@ -81,7 +80,7 @@ impl Validators {
           if sessions.get(&network) == Some(&session) {
             Ok(None)
           } else {
-            match temporal_serai.current_validators(network.into()).await {
+            match serai.current_validators(network.into()).await {
               Ok(Some(validators)) => Ok(Some((
                 network,
                 session,

coordinator/src/db.rs

@@ -50,7 +50,6 @@ fn tributary_db_folder(set: ExternalValidatorSet) -> String {
     ExternalNetworkId::Bitcoin => "Bitcoin",
     ExternalNetworkId::Ethereum => "Ethereum",
     ExternalNetworkId::Monero => "Monero",
-    _ => panic!("unrecognized `ExternalNetworkId`"),
   };
   format!("{root_path}/tributary-{network}-{}", set.session.0)
 }

coordinator/substrate/Cargo.toml

@@ -29,7 +29,6 @@ serai-client-serai = { path = "../../substrate/client/serai", default-features =
 log = { version = "0.4", default-features = false, features = ["std"] }
 
 futures = { version = "0.3", default-features = false, features = ["std"] }
-tokio = { version = "1", default-features = false }
 
 serai-db = { path = "../../common/db", version = "0.1.1" }
 serai-task = { path = "../../common/task", version = "0.1" }

coordinator/substrate/src/canonical.rs

@@ -73,21 +73,13 @@ impl<D: Db> ContinuallyRan for CanonicalEventStream<D> {
               }
               Err(serai_cosign::Faulted) => return Err("cosigning process faulted".to_string()),
             };
-            let temporal_serai = serai.as_of(block_hash).await.map_err(|e| format!("{e}"))?;
+            let events = serai.events(block_hash).await.map_err(|e| format!("{e}"))?;
-            let temporal_serai_validators = temporal_serai.validator_sets();
+            let set_keys_events = events.validator_sets().set_keys_events().cloned().collect();
-            let temporal_serai_instructions = temporal_serai.in_instructions();
+            let slash_report_events =
-            let temporal_serai_coins = temporal_serai.coins();
+              events.validator_sets().slash_report_events().cloned().collect();
-
+            let batch_events = events.in_instructions().batch_events().cloned().collect();
-            let (block, set_keys_events, slash_report_events, batch_events, burn_events) =
+            let burn_events = events.coins().burn_with_instruction_events().cloned().collect();
-              tokio::try_join!(
+            let Some(block) = serai.block(block_hash).await.map_err(|e| format!("{e:?}"))? else {
-                serai.block(block_hash),
-                temporal_serai_validators.set_keys_events(),
-                temporal_serai_validators.slash_report_events(),
-                temporal_serai_instructions.batch_events(),
-                temporal_serai_coins.burn_with_instruction_events(),
-              )
-              .map_err(|e| format!("{e:?}"))?;
-            let Some(block) = block else {
               Err(format!("Serai node didn't have cosigned block #{block_number}"))?
             };
 

coordinator/substrate/src/ephemeral.rs

@@ -6,7 +6,7 @@ use futures::stream::{StreamExt, FuturesOrdered};
 use serai_client_serai::{
   abi::primitives::{
     BlockHash,
-    crypto::EmbeddedEllipticCurveKeys,
+    crypto::EmbeddedEllipticCurveKeys as EmbeddedEllipticCurveKeysStruct,
     network_id::ExternalNetworkId,
     validator_sets::{KeyShares, ExternalValidatorSet},
     address::SeraiAddress,
@@ -24,6 +24,10 @@ use crate::NewSetInformation;
 create_db!(
   CoordinatorSubstrateEphemeral {
     NextBlock: () -> u64,
+    EmbeddedEllipticCurveKeys: (
+      network: ExternalNetworkId,
+      validator: SeraiAddress
+    ) -> EmbeddedEllipticCurveKeysStruct,
   }
 );
 
@@ -56,6 +60,7 @@ impl<D: Db> ContinuallyRan for EphemeralEventStream<D> {
       struct EphemeralEvents {
         block_hash: BlockHash,
         time: u64,
+        embedded_elliptic_curve_keys_events: Vec<serai_client_serai::abi::validator_sets::Event>,
         set_decided_events: Vec<serai_client_serai::abi::validator_sets::Event>,
         accepted_handover_events: Vec<serai_client_serai::abi::validator_sets::Event>,
       }
@@ -76,15 +81,17 @@ impl<D: Db> ContinuallyRan for EphemeralEventStream<D> {
               Err(serai_cosign::Faulted) => return Err("cosigning process faulted".to_string()),
             };
 
-            let temporal_serai = serai.as_of(block_hash).await.map_err(|e| format!("{e}"))?;
+            let events = serai.events(block_hash).await.map_err(|e| format!("{e}"))?;
-            let temporal_serai_validators = temporal_serai.validator_sets();
+            let embedded_elliptic_curve_keys_events = events
-            let (block, set_decided_events, accepted_handover_events) = tokio::try_join!(
+              .validator_sets()
-              serai.block(block_hash),
+              .set_embedded_elliptic_curve_keys_events()
-              temporal_serai_validators.set_decided_events(),
+              .cloned()
-              temporal_serai_validators.accepted_handover_events(),
+              .collect::<Vec<_>>();
-            )
+            let set_decided_events =
-            .map_err(|e| format!("{e:?}"))?;
+              events.validator_sets().set_decided_events().cloned().collect::<Vec<_>>();
-            let Some(block) = block else {
+            let accepted_handover_events =
+              events.validator_sets().accepted_handover_events().cloned().collect::<Vec<_>>();
+            let Some(block) = serai.block(block_hash).await.map_err(|e| format!("{e:?}"))? else {
               Err(format!("Serai node didn't have cosigned block #{block_number}"))?
             };
 
@@ -92,7 +99,13 @@ impl<D: Db> ContinuallyRan for EphemeralEventStream<D> {
             let time = block.header.unix_time_in_millis() / 1000;
             Ok((
               block_number,
-              EphemeralEvents { block_hash, time, set_decided_events, accepted_handover_events },
+              EphemeralEvents {
+                block_hash,
+                time,
+                embedded_elliptic_curve_keys_events,
+                set_decided_events,
+                accepted_handover_events,
+              },
             ))
           }
         }
@@ -123,30 +136,39 @@ impl<D: Db> ContinuallyRan for EphemeralEventStream<D> {
 
         let mut txn = self.db.txn();
 
+        for event in block.embedded_elliptic_curve_keys_events {
+          let serai_client_serai::abi::validator_sets::Event::SetEmbeddedEllipticCurveKeys {
+            validator,
+            keys,
+          } = &event
+          else {
+            panic!(
+              "{}: {event:?}",
+              "`SetEmbeddedEllipticCurveKeys` event wasn't a `SetEmbeddedEllipticCurveKeys` event"
+            );
+          };
+
+          EmbeddedEllipticCurveKeys::set(&mut txn, keys.network(), *validator, keys);
+        }
+
         for set_decided in block.set_decided_events {
           let serai_client_serai::abi::validator_sets::Event::SetDecided { set, validators } =
             &set_decided
           else {
             panic!("`SetDecided` event wasn't a `SetDecided` event: {set_decided:?}");
           };
+
           // We only coordinate over external networks
           let Ok(set) = ExternalValidatorSet::try_from(*set) else { continue };
-
-          let serai = self.serai.as_of(block.block_hash).await.map_err(|e| format!("{e}"))?;
-          let serai = serai.validator_sets();
           let validators =
-            validators.iter().map(|(validator, weight)| (*validator, weight)).collect::<Vec<_>>();
+            validators.iter().map(|(validator, weight)| (*validator, weight.0)).collect::<Vec<_>>();
+
           let in_set = validators.iter().any(|(validator, _)| *validator == self.validator);
           if in_set {
             if u16::try_from(validators.len()).is_err() {
               Err("more than u16::MAX validators sent")?;
             }
 
-            let validators = validators
-              .into_iter()
-              .map(|(validator, weight)| (validator, weight.0))
-              .collect::<Vec<_>>();
-
             // Do the summation in u32 so we don't risk a u16 overflow
             let total_weight = validators.iter().map(|(_, weight)| u32::from(*weight)).sum::<u32>();
             if total_weight > u32::from(KeyShares::MAX_PER_SET) {
@@ -159,50 +181,26 @@ impl<D: Db> ContinuallyRan for EphemeralEventStream<D> {
               .expect("value smaller than `u16` constant but doesn't fit in `u16`");
 
             // Fetch all of the validators' embedded elliptic curve keys
-            let mut embedded_elliptic_curve_keys = FuturesOrdered::new();
-            for (validator, _) in &validators {
-              let validator = *validator;
-              // try_join doesn't return a future so we need to wrap it in this additional async
-              // block
-              embedded_elliptic_curve_keys.push_back({
-                let serai = serai.clone();
-                async move {
-                  match serai.embedded_elliptic_curve_keys(validator, set.network).await {
-                    Ok(Some(keys)) => Ok(Some((
-                      validator,
-                      match keys {
-                        EmbeddedEllipticCurveKeys::Bitcoin(substrate, external) => {
-                          assert_eq!(set.network, ExternalNetworkId::Bitcoin);
-                          (substrate, external.as_slice().to_vec())
-                        }
-                        EmbeddedEllipticCurveKeys::Ethereum(substrate, external) => {
-                          assert_eq!(set.network, ExternalNetworkId::Ethereum);
-                          (substrate, external.as_slice().to_vec())
-                        }
-                        EmbeddedEllipticCurveKeys::Monero(substrate) => {
-                          assert_eq!(set.network, ExternalNetworkId::Monero);
-                          (substrate, substrate.as_slice().to_vec())
-                        }
-                      },
-                    ))),
-                    Ok(None) => Ok(None),
-                    Err(e) => Err(e),
-                  }
-                }
-              });
-            }
-
             let mut evrf_public_keys = Vec::with_capacity(usize::from(total_weight));
             for (validator, weight) in &validators {
-              let Some((future_validator, (substrate_embedded_key, external_embedded_key))) =
+              let keys = match EmbeddedEllipticCurveKeys::get(&txn, set.network, *validator)
-                embedded_elliptic_curve_keys.next().await.unwrap().map_err(|e| format!("{e:?}"))?
+                .expect("selected validator lacked embedded elliptic curve keys")
-              else {
+              {
-                Err("`SetDecided` with validator missing an embedded key".to_string())?
+                EmbeddedEllipticCurveKeysStruct::Bitcoin(substrate, external) => {
+                  assert_eq!(set.network, ExternalNetworkId::Bitcoin);
+                  (substrate, external.to_vec())
+                }
+                EmbeddedEllipticCurveKeysStruct::Ethereum(substrate, external) => {
+                  assert_eq!(set.network, ExternalNetworkId::Ethereum);
+                  (substrate, external.to_vec())
+                }
+                EmbeddedEllipticCurveKeysStruct::Monero(substrate) => {
+                  assert_eq!(set.network, ExternalNetworkId::Monero);
+                  (substrate, substrate.to_vec())
+                }
               };
-              assert_eq!(*validator, future_validator);
-
               for _ in 0 .. *weight {
-                evrf_public_keys.push((substrate_embedded_key, external_embedded_key.clone()));
+                evrf_public_keys.push(keys.clone());
               }
             }
 
@@ -213,6 +211,7 @@ impl<D: Db> ContinuallyRan for EphemeralEventStream<D> {
               // TODO: This should be inlined into the Processor's key gen code
               // It's legacy from when we removed participants from the key gen
               threshold: ((total_weight * 2) / 3) + 1,
+              // TODO: Why are `validators` and `evrf_public_keys` two separate fields?
               validators,
               evrf_public_keys,
               participant_indexes: Default::default(),

coordinator/substrate/src/publish_slash_report.rs

@@ -36,8 +36,7 @@ impl<D: Db> PublishSlashReportTask<D> {
 
     // This uses the latest finalized block, not the latest cosigned block, which should be
     // fine as in the worst case, the only impact is no longer attempting TX publication
-    let serai = self.serai.as_of_latest_finalized_block().await.map_err(|e| format!("{e:?}"))?;
+    let serai = self.serai.state().await.map_err(|e| format!("{e:?}"))?;
-    let serai = serai.validator_sets();
     let session_after_slash_report = Session(session.0 + 1);
     let current_session =
       serai.current_session(network.into()).await.map_err(|e| format!("{e:?}"))?;

coordinator/substrate/src/set_keys.rs

@@ -40,9 +40,7 @@ impl<D: Db> ContinuallyRan for SetKeysTask<D> {
 
         // This uses the latest finalized block, not the latest cosigned block, which should be
         // fine as in the worst case, the only impact is no longer attempting TX publication
-        let serai =
+        let serai = self.serai.state().await.map_err(|e| format!("{e:?}"))?;
-          self.serai.as_of_latest_finalized_block().await.map_err(|e| format!("{e:?}"))?;
-        let serai = serai.validator_sets();
         let current_session =
           serai.current_session(network.into()).await.map_err(|e| format!("{e:?}"))?;
         let current_session = current_session.map(|session| session.0);

message-queue/src/main.rs

@@ -219,7 +219,6 @@ async fn main() {
       ExternalNetworkId::Bitcoin => "BITCOIN_KEY",
       ExternalNetworkId::Ethereum => "ETHEREUM_KEY",
       ExternalNetworkId::Monero => "MONERO_KEY",
-      _ => panic!("unrecognized network"),
     }) else {
       continue;
     };

substrate/abi/src/validator_sets.rs

@@ -3,7 +3,7 @@ use alloc::vec::Vec;
 use borsh::{BorshSerialize, BorshDeserialize};
 
 use serai_primitives::{
-  crypto::{SignedEmbeddedEllipticCurveKeys, KeyPair, Signature},
+  crypto::{EmbeddedEllipticCurveKeys, SignedEmbeddedEllipticCurveKeys, KeyPair, Signature},
   address::SeraiAddress,
   balance::Amount,
   network_id::*,
@@ -125,8 +125,8 @@ pub enum Event {
   SetEmbeddedEllipticCurveKeys {
     /// The validator which set their keys.
     validator: SeraiAddress,
-    /// The network which they set embedded elliptic curve keys for.
+    /// The embedded elliptic curve keys they set.
-    network: ExternalNetworkId,
+    keys: EmbeddedEllipticCurveKeys,
   },
   /// A validator's allocation to a network has increased.
   Allocation {

substrate/client/serai/Cargo.toml

@@ -28,8 +28,6 @@ borsh = { version = "1", default-features = false, features = ["std"] }
 bitvec = { version = "1", default-features = false, features = ["alloc", "std"] }
 serai-abi = { path = "../../abi", version = "0.1", default-features = false, features = ["std"] }
 
-async-lock = "3"
-
 [dev-dependencies]
 blake2 = { version = "0.11.0-rc.3", default-features = false }
 sp-core = { git = "https://github.com/serai-dex/patch-polkadot-sdk", rev = "e01101b68c5b0f588dd4cdee48f801a2c1f75b84" }

substrate/client/serai/src/coins.rs

@@ -1,76 +1,37 @@
 pub use serai_abi::coins::Event;
 
-use crate::{RpcError, TemporalSerai};
+use crate::{RpcError, Events};
 
-/// A `TemporalSerai` scoped to the coins module.
+/// An `Events` scoped to the coins module.
 #[derive(Clone)]
-pub struct Coins<'a>(pub(super) &'a TemporalSerai<'a>);
+pub struct Coins(pub(super) Events);
 
-impl<'a> Coins<'a> {
+impl Coins {
   /// The events from the coins module.
-  pub async fn events(&self) -> Result<Vec<Event>, RpcError> {
+  pub fn events(&self) -> impl Iterator<Item = &Event> {
-    Ok(
+    self.0.events().flatten().filter_map(|event| match event {
-      self
+      serai_abi::Event::Coins(event) => Some(event),
-        .0
-        .events_borrowed()
-        .await?
-        .as_ref()
-        .expect("`TemporalSerai::events` returned None")
-        .iter()
-        .flat_map(IntoIterator::into_iter)
-        .filter_map(|event| match event {
-          serai_abi::Event::Coins(event) => Some(event.clone()),
       _ => None,
     })
-        .collect(),
-    )
   }
 
   /// The `Mint` events from the coins module.
-  pub async fn mint_events(&self) -> Result<Vec<Event>, RpcError> {
+  pub fn mint_events(&self) -> impl Iterator<Item = &Event> {
-    Ok(
+    self.events().filter(|event| matches!(event, Event::Mint { .. }))
-      self
-        .events()
-        .await?
-        .into_iter()
-        .filter(|event| matches!(event, Event::Mint { .. }))
-        .collect(),
-    )
   }
 
   /// The `Transfer` events from the coins module.
-  pub async fn transfer_events(&self) -> Result<Vec<Event>, RpcError> {
+  pub fn transfer_events(&self) -> impl Iterator<Item = &Event> {
-    Ok(
+    self.events().filter(|event| matches!(event, Event::Transfer { .. }))
-      self
-        .events()
-        .await?
-        .into_iter()
-        .filter(|event| matches!(event, Event::Transfer { .. }))
-        .collect(),
-    )
   }
 
   /// The `Burn` events from the coins module.
-  pub async fn burn_events(&self) -> Result<Vec<Event>, RpcError> {
+  pub fn burn_events(&self) -> impl Iterator<Item = &Event> {
-    Ok(
+    self.events().filter(|event| matches!(event, Event::Burn { .. }))
-      self
-        .events()
-        .await?
-        .into_iter()
-        .filter(|event| matches!(event, Event::Burn { .. }))
-        .collect(),
-    )
   }
 
   /// The `BurnWithInstruction` events from the coins module.
-  pub async fn burn_with_instruction_events(&self) -> Result<Vec<Event>, RpcError> {
+  pub fn burn_with_instruction_events(&self) -> impl Iterator<Item = &Event> {
-    Ok(
+    self.events().filter(|event| matches!(event, Event::BurnWithInstruction { .. }))
-      self
-        .events()
-        .await?
-        .into_iter()
-        .filter(|event| matches!(event, Event::BurnWithInstruction { .. }))
-        .collect(),
-    )
   }
 }

substrate/client/serai/src/in_instructions.rs

@@ -4,42 +4,24 @@ pub use serai_abi::{
   UnsignedCall, Transaction,
 };
 
-use crate::{RpcError, TemporalSerai};
+use crate::{RpcError, Events};
 
-/// A `TemporalSerai` scoped to the in instructions module.
+/// An `Events` scoped to the in instructions module.
 #[derive(Clone)]
-pub struct InInstructions<'serai>(pub(super) &'serai TemporalSerai<'serai>);
+pub struct InInstructions(pub(super) Events);
 
-impl<'serai> InInstructions<'serai> {
+impl InInstructions {
   /// The events from the in instructions module.
-  pub async fn events(&self) -> Result<Vec<Event>, RpcError> {
+  fn events(&self) -> impl Iterator<Item = &Event> {
-    Ok(
+    self.0.events().flatten().filter_map(|event| match event {
-      self
+      serai_abi::Event::InInstructions(event) => Some(event),
-        .0
-        .events_borrowed()
-        .await?
-        .as_ref()
-        .expect("`TemporalSerai::events` returned None")
-        .iter()
-        .flat_map(IntoIterator::into_iter)
-        .filter_map(|event| match event {
-          serai_abi::Event::InInstructions(event) => Some(event.clone()),
       _ => None,
     })
-        .collect(),
-    )
   }
 
   /// The `Batch` events from the in instructions module.
-  pub async fn batch_events(&self) -> Result<Vec<Event>, RpcError> {
+  pub fn batch_events(&self) -> impl Iterator<Item = &Event> {
-    Ok(
+    self.events().filter(|event| matches!(event, Event::Batch { .. }))
-      self
-        .events()
-        .await?
-        .into_iter()
-        .filter(|event| matches!(event, Event::Batch { .. }))
-        .collect(),
-    )
   }
 
   /// Create a transaction to execute a batch.

substrate/client/serai/src/lib.rs

@@ -17,8 +17,6 @@ use abi::{
   Transaction, Block, Event,
 };
 
-use async_lock::RwLock;
-
 mod coins;
 pub use coins::Coins;
 
@@ -55,16 +53,18 @@ pub struct Serai {
   client: TokioClient,
 }
 
-/// An RPC client to a Serai node, scoped to a specific block.
+/// The events from a specific block.
-///
-/// Upon any request being made for the events emitted by this block, the entire list of events
-/// from this block will be cached within this. This allows future calls for events to be done
-/// cheaply.
 #[derive(Clone)]
-pub struct TemporalSerai<'serai> {
+pub struct Events {
+  // These are cached within an `Arc` for cheap `Clone`s
+  events: Arc<Vec<Vec<Event>>>,
+}
+
+/// An RPC client to a Serai node, scoped to a specific block.
+#[derive(Clone)]
+pub struct State<'serai> {
   serai: &'serai Serai,
   block: BlockHash,
-  events: Arc<RwLock<Option<Vec<Vec<Event>>>>>,
 }
 
 impl Serai {
@@ -175,70 +175,12 @@ impl Serai {
       .await
   }
 
-  /// Scope this RPC client to the state as of a specific block.
+  /// Fetch the events of a specific block.
-  ///
+  pub async fn events(&self, block: BlockHash) -> Result<Events, RpcError> {
-  /// This will yield an error if the block chosen isn't finalized. This ensures, given an honest
+    Ok(Events {
-  /// node, that this scope will be available for the lifetime of this object.
+      events: Arc::new(
-  pub async fn as_of(&self, block: BlockHash) -> Result<TemporalSerai<'_>, RpcError> {
-    if !self.finalized(block).await? {
-      Err(RpcError::NotFinalized)?;
-    }
-    Ok(TemporalSerai { serai: self, block, events: Arc::new(RwLock::new(None)) })
-  }
-
-  /// Scope this RPC client to the state as of the latest finalized block.
-  pub async fn as_of_latest_finalized_block(&self) -> Result<TemporalSerai<'_>, RpcError> {
-    let block = self
-      .block_by_number(self.latest_finalized_block_number().await?)
-      .await?
-      .ok_or_else(|| RpcError::InvalidNode("couldn't fetch latest finalized block".to_string()))?;
-    Ok(TemporalSerai {
-      serai: self,
-      block: block.header.hash(),
-      events: Arc::new(RwLock::new(None)),
-    })
-  }
-
-  /// Return the P2P addresses for the validators of the specified network.
-  pub async fn p2p_validators(&self, network: ExternalNetworkId) -> Result<Vec<String>, RpcError> {
         self
-      .call(
+          .call::<Vec<Vec<String>>>("blockchain/events", &format!(r#"{{ "block": "{block}" }}"#))
-        "p2p_validators",
-        match network {
-          ExternalNetworkId::Bitcoin => r#"["bitcoin"]"#,
-          ExternalNetworkId::Ethereum => r#"["ethereum"]"#,
-          ExternalNetworkId::Monero => r#"["monero"]"#,
-          _ => Err(RpcError::InternalError("unrecognized external network ID".to_string()))?,
-        },
-      )
-      .await
-  }
-}
-
-impl<'serai> TemporalSerai<'serai> {
-  async fn call<ResponseValue: Default + JsonDeserialize>(
-    &self,
-    method: &str,
-    params: &str,
-  ) -> Result<ResponseValue, RpcError> {
-    self.serai.call(method, &format!(r#"{{ "block": "{}" {params} }}"#, self.block)).await
-  }
-
-  /// Fetch the events for this block.
-  ///
-  /// The returned `Option` will always be `Some(_)`.
-  async fn events_borrowed(
-    &self,
-  ) -> Result<async_lock::RwLockReadGuard<'_, Option<Vec<Vec<Event>>>>, RpcError> {
-    let mut events = self.events.read().await;
-    if events.is_none() {
-      drop(events);
-      {
-        let mut events_mut = self.events.write().await;
-        if events_mut.is_none() {
-          *events_mut = Some(
-            self
-              .call::<Vec<Vec<String>>>("blockchain/events", "")
           .await?
           .into_iter()
           .map(|events_per_tx| {
@@ -257,34 +199,65 @@ impl<'serai> TemporalSerai<'serai> {
               .collect::<Result<Vec<_>, _>>()
           })
           .collect::<Result<Vec<_>, _>>()?,
-          );
+      ),
-        }
+    })
-      }
-      events = self.events.read().await;
-    }
-    Ok(events)
   }
 
-  /// Fetch the events for this block.
+  /// Scope this RPC client to the state as of the latest finalized block.
+  pub async fn state(&self) -> Result<State<'_>, RpcError> {
+    let block = self
+      .block_by_number(self.latest_finalized_block_number().await?)
+      .await?
+      .ok_or_else(|| RpcError::InvalidNode("couldn't fetch latest finalized block".to_string()))?;
+    Ok(State { serai: self, block: block.header.hash() })
+  }
+
+  /// Return the P2P addresses for the validators of the specified network.
+  pub async fn p2p_validators(&self, network: ExternalNetworkId) -> Result<Vec<String>, RpcError> {
+    self
+      .call(
+        "p2p_validators",
+        match network {
+          ExternalNetworkId::Bitcoin => r#"["bitcoin"]"#,
+          ExternalNetworkId::Ethereum => r#"["ethereum"]"#,
+          ExternalNetworkId::Monero => r#"["monero"]"#,
+        },
+      )
+      .await
+  }
+}
+
+impl Events {
+  /// The events within this container.
   ///
-  /// These will be grouped by the transactions which emitted them, including the inherent
+  /// This will yield the events for each transaction within the block, including the implicit
-  /// transactions at the start and end of every block.
+  /// transactions at the start and end of each block, within an outer container.
-  pub async fn events(&self) -> Result<Vec<Vec<Event>>, RpcError> {
+  pub fn events(&self) -> impl Iterator<Item: IntoIterator<Item = &Event>> {
-    Ok(self.events_borrowed().await?.clone().expect("`TemporalSerai::events` returned None"))
+    self.events.iter()
   }
 
   /// Scope to the coins module.
-  pub fn coins(&self) -> Coins<'_> {
+  pub fn coins(&self) -> Coins {
-    Coins(self)
+    Coins(self.clone())
   }
 
   /// Scope to the validator sets module.
-  pub fn validator_sets(&self) -> ValidatorSets<'_> {
+  pub fn validator_sets(&self) -> ValidatorSets {
-    ValidatorSets(self)
+    ValidatorSets(self.clone())
   }
 
   /// Scope to the in instructions module.
-  pub fn in_instructions(&self) -> InInstructions<'_> {
+  pub fn in_instructions(&self) -> InInstructions {
-    InInstructions(self)
+    InInstructions(self.clone())
+  }
+}
+
+impl<'serai> State<'serai> {
+  async fn call<ResponseValue: Default + JsonDeserialize>(
+    &self,
+    method: &str,
+    params: &str,
+  ) -> Result<ResponseValue, RpcError> {
+    self.serai.call(method, &format!(r#"{{ "block": "{}" {params} }}"#, self.block)).await
   }
 }

substrate/client/serai/src/validator_sets.rs

@@ -14,203 +14,68 @@ pub use serai_abi::{
   UnsignedCall, Transaction,
 };
 
-use crate::{RpcError, TemporalSerai};
+use crate::{RpcError, Events, State};
 
-fn rpc_network(network: impl Into<NetworkId>) -> Result<&'static str, RpcError> {
+fn rpc_network(network: impl Into<NetworkId>) -> &'static str {
-  Ok(match network.into() {
+  match network.into() {
     NetworkId::Serai => r#""serai""#,
     NetworkId::External(ExternalNetworkId::Bitcoin) => r#""bitcoin""#,
     NetworkId::External(ExternalNetworkId::Ethereum) => r#""ethereum""#,
     NetworkId::External(ExternalNetworkId::Monero) => r#""monero""#,
-    _ => Err(RpcError::InternalError("unrecognized network ID".to_string()))?,
+  }
-  })
 }
 
-/// A `TemporalSerai` scoped to the validator sets module.
+/// An `Events` scoped to the validator sets module.
 #[derive(Clone)]
-pub struct ValidatorSets<'serai>(pub(super) &'serai TemporalSerai<'serai>);
+pub struct ValidatorSets(pub(super) Events);
 
-impl<'serai> ValidatorSets<'serai> {
+impl ValidatorSets {
   /// The events from the validator sets module.
-  pub async fn events(&self) -> Result<Vec<Event>, RpcError> {
+  pub fn events(&self) -> impl Iterator<Item = &Event> {
-    Ok(
+    self.0.events().flatten().filter_map(|event| match event {
-      self
+      serai_abi::Event::ValidatorSets(event) => Some(event),
-        .0
-        .events_borrowed()
-        .await?
-        .as_ref()
-        .expect("`TemporalSerai::events` returned None")
-        .iter()
-        .flat_map(IntoIterator::into_iter)
-        .filter_map(|event| match event {
-          serai_abi::Event::ValidatorSets(event) => Some(event.clone()),
       _ => None,
     })
-        .collect(),
-    )
   }
 
   /// The `SetDecided` events from the validator sets module.
-  pub async fn set_decided_events(&self) -> Result<Vec<Event>, RpcError> {
+  pub fn set_decided_events(&self) -> impl Iterator<Item = &Event> {
-    Ok(
+    self.events().filter(|event| matches!(event, Event::SetDecided { .. }))
-      self
-        .events()
-        .await?
-        .into_iter()
-        .filter(|event| matches!(event, Event::SetDecided { .. }))
-        .collect(),
-    )
   }
 
   /// The `SetKeys` events from the validator sets module.
-  pub async fn set_keys_events(&self) -> Result<Vec<Event>, RpcError> {
+  pub fn set_keys_events(&self) -> impl Iterator<Item = &Event> {
-    Ok(
+    self.events().filter(|event| matches!(event, Event::SetKeys { .. }))
-      self
-        .events()
-        .await?
-        .into_iter()
-        .filter(|event| matches!(event, Event::SetKeys { .. }))
-        .collect(),
-    )
   }
 
   /// The `AcceptedHandover` events from the validator sets module.
-  pub async fn accepted_handover_events(&self) -> Result<Vec<Event>, RpcError> {
+  pub fn accepted_handover_events(&self) -> impl Iterator<Item = &Event> {
-    Ok(
+    self.events().filter(|event| matches!(event, Event::AcceptedHandover { .. }))
-      self
-        .events()
-        .await?
-        .into_iter()
-        .filter(|event| matches!(event, Event::AcceptedHandover { .. }))
-        .collect(),
-    )
   }
 
   /// The `SlashReport` events from the validator sets module.
-  pub async fn slash_report_events(&self) -> Result<Vec<Event>, RpcError> {
+  pub fn slash_report_events(&self) -> impl Iterator<Item = &Event> {
-    Ok(
+    self.events().filter(|event| matches!(event, Event::SlashReport { .. }))
-      self
-        .events()
-        .await?
-        .into_iter()
-        .filter(|event| matches!(event, Event::SlashReport { .. }))
-        .collect(),
-    )
   }
 
-  /// The current session for the specified network.
+  /// The `SetEmbeddedEllipticCurveKeys` events from the validator sets module.
-  pub async fn current_session(&self, network: NetworkId) -> Result<Option<Session>, RpcError> {
+  pub fn set_embedded_elliptic_curve_keys_events(&self) -> impl Iterator<Item = &Event> {
-    Ok(
+    self.events().filter(|event| matches!(event, Event::SetEmbeddedEllipticCurveKeys { .. }))
-      self
-        .0
-        .call::<Option<_>>(
-          "validator-sets/current_session",
-          &format!(r#", "network": {} "#, rpc_network(network)?),
-        )
-        .await?
-        .map(Session),
-    )
   }
 
-  /// The stake for the current validators for the specified network.
+  /// The `Allocation` events from the validator sets module.
-  pub async fn current_stake(&self, network: NetworkId) -> Result<Option<Amount>, RpcError> {
+  pub fn allocation_events(&self) -> impl Iterator<Item = &Event> {
-    Ok(
+    self.events().filter(|event| matches!(event, Event::Allocation { .. }))
-      self
-        .0
-        .call::<Option<_>>(
-          "validator-sets/current_stake",
-          &format!(r#", "network": {} "#, rpc_network(network)?),
-        )
-        .await?
-        .map(Amount),
-    )
   }
 
-  /// The keys for the specified validator set.
+  /// The `Deallocation` events from the validator sets module.
-  pub async fn keys(&self, set: ExternalValidatorSet) -> Result<Option<KeyPair>, RpcError> {
+  pub fn deallocation_events(&self) -> impl Iterator<Item = &Event> {
-    let Some(key_pair) = self
+    self.events().filter(|event| matches!(event, Event::Deallocation { .. }))
-      .0
-      .call::<Option<String>>(
-        "validator-sets/keys",
-        &format!(r#", "network": {}, "session": {} "#, rpc_network(set.network)?, set.session.0),
-      )
-      .await?
-    else {
-      return Ok(None);
-    };
-    KeyPair::deserialize(
-      &mut hex::decode(key_pair)
-        .map_err(|_| RpcError::InvalidNode("validator set's keys weren't valid hex".to_string()))?
-        .as_slice(),
-    )
-    .map(Some)
-    .map_err(|_| RpcError::InvalidNode("validator set's keys weren't a valid key pair".to_string()))
   }
 
-  /// The current validators for the specified network.
+  /// The `DelayedDeallocationClaimed` events from the validator sets module.
-  pub async fn current_validators(
+  pub fn delayed_deallocation_claimed_events(&self) -> impl Iterator<Item = &Event> {
-    &self,
+    self.events().filter(|event| matches!(event, Event::DelayedDeallocationClaimed { .. }))
-    network: NetworkId,
-  ) -> Result<Option<Vec<SeraiAddress>>, RpcError> {
-    self
-      .0
-      .call::<Option<Vec<String>>>(
-        "validator-sets/current_validators",
-        &format!(r#", "network": {} "#, rpc_network(network)?),
-      )
-      .await?
-      .map(|validators| {
-        validators
-          .into_iter()
-          .map(|addr| {
-            SeraiAddress::from_str(&addr)
-              .map_err(|_| RpcError::InvalidNode("validator's address was invalid".to_string()))
-          })
-          .collect()
-      })
-      .transpose()
-  }
-
-  /// If the prior validators for this network is still expected to publish a slash report.
-  pub async fn pending_slash_report(&self, network: ExternalNetworkId) -> Result<bool, RpcError> {
-    self
-      .0
-      .call(
-        "validator-sets/pending_slash_report",
-        &format!(r#", "network": {} "#, rpc_network(network)?),
-      )
-      .await
-  }
-
-  /// The key on an embedded elliptic curve for the specified validator.
-  pub async fn embedded_elliptic_curve_keys(
-    &self,
-    validator: SeraiAddress,
-    network: ExternalNetworkId,
-  ) -> Result<Option<EmbeddedEllipticCurveKeys>, RpcError> {
-    let Some(keys) = self
-      .0
-      .call::<Option<String>>(
-        "validator-sets/embedded_elliptic_curve_keys",
-        &format!(r#", "validator": {validator}, "network": {} "#, rpc_network(network)?),
-      )
-      .await?
-    else {
-      return Ok(None);
-    };
-    EmbeddedEllipticCurveKeys::deserialize(
-      &mut hex::decode(keys)
-        .map_err(|_| {
-          RpcError::InvalidNode(
-            "validator's embedded elliptic curve keys weren't valid hex".to_string(),
-          )
-        })?
-        .as_slice(),
-    )
-    .map(Some)
-    .map_err(|_| {
-      RpcError::InvalidNode("validator's embedded elliptic curve keys weren't valid".to_string())
-    })
   }
 
   /// Create a transaction to set a validator set's keys.
@@ -247,3 +112,114 @@ impl<'serai> ValidatorSets<'serai> {
     }
   }
 }
+
+impl<'serai> State<'serai> {
+  /// The current session for the specified network.
+  pub async fn current_session(&self, network: NetworkId) -> Result<Option<Session>, RpcError> {
+    Ok(
+      self
+        .call::<Option<_>>(
+          "validator-sets/current_session",
+          &format!(r#", "network": {} "#, rpc_network(network)),
+        )
+        .await?
+        .map(Session),
+    )
+  }
+
+  /// The stake for the current validators for the specified network.
+  pub async fn current_stake(&self, network: NetworkId) -> Result<Option<Amount>, RpcError> {
+    Ok(
+      self
+        .call::<Option<_>>(
+          "validator-sets/current_stake",
+          &format!(r#", "network": {} "#, rpc_network(network)),
+        )
+        .await?
+        .map(Amount),
+    )
+  }
+
+  /// The keys for the specified validator set.
+  pub async fn keys(&self, set: ExternalValidatorSet) -> Result<Option<KeyPair>, RpcError> {
+    let Some(key_pair) = self
+      .call::<Option<String>>(
+        "validator-sets/keys",
+        &format!(r#", "network": {}, "session": {} "#, rpc_network(set.network), set.session.0),
+      )
+      .await?
+    else {
+      return Ok(None);
+    };
+    KeyPair::deserialize(
+      &mut hex::decode(key_pair)
+        .map_err(|_| RpcError::InvalidNode("validator set's keys weren't valid hex".to_string()))?
+        .as_slice(),
+    )
+    .map(Some)
+    .map_err(|_| RpcError::InvalidNode("validator set's keys weren't a valid key pair".to_string()))
+  }
+
+  /// The current validators for the specified network.
+  pub async fn current_validators(
+    &self,
+    network: NetworkId,
+  ) -> Result<Option<Vec<SeraiAddress>>, RpcError> {
+    self
+      .call::<Option<Vec<String>>>(
+        "validator-sets/current_validators",
+        &format!(r#", "network": {} "#, rpc_network(network)),
+      )
+      .await?
+      .map(|validators| {
+        validators
+          .into_iter()
+          .map(|addr| {
+            SeraiAddress::from_str(&addr)
+              .map_err(|_| RpcError::InvalidNode("validator's address was invalid".to_string()))
+          })
+          .collect()
+      })
+      .transpose()
+  }
+
+  /// If the prior validators for this network is still expected to publish a slash report.
+  pub async fn pending_slash_report(&self, network: ExternalNetworkId) -> Result<bool, RpcError> {
+    self
+      .call(
+        "validator-sets/pending_slash_report",
+        &format!(r#", "network": {} "#, rpc_network(network)),
+      )
+      .await
+  }
+
+  /// The key on an embedded elliptic curve for the specified validator.
+  pub async fn embedded_elliptic_curve_keys(
+    &self,
+    validator: SeraiAddress,
+    network: ExternalNetworkId,
+  ) -> Result<Option<EmbeddedEllipticCurveKeys>, RpcError> {
+    let Some(keys) = self
+      .call::<Option<String>>(
+        "validator-sets/embedded_elliptic_curve_keys",
+        &format!(r#", "validator": "{validator}", "network": {} "#, rpc_network(network)),
+      )
+      .await?
+    else {
+      return Ok(None);
+    };
+    EmbeddedEllipticCurveKeys::deserialize(
+      &mut hex::decode(keys)
+        .map_err(|_| {
+          RpcError::InvalidNode(
+            "validator's embedded elliptic curve keys weren't valid hex".to_string(),
+          )
+        })?
+        .as_slice(),
+    )
+    .map(Some)
+    .map_err(|_| {
+      RpcError::InvalidNode("validator's embedded elliptic curve keys weren't valid".to_string())
+    })
+  }
+}

substrate/client/serai/tests/blockchain.rs

@@ -133,7 +133,11 @@ async fn blockchain() {
               .chain(block.transactions.iter().map(serai_abi::Transaction::hash))
               .chain(core::iter::once(end_transaction));
 
-            let events = serai.as_of(block.header.hash()).await.unwrap().events().await.unwrap();
+            let events = serai.events(block.header.hash()).await.unwrap();
+            let events = events
+              .events()
+              .map(|iter| iter.into_iter().cloned().collect::<Vec<_>>())
+              .collect::<Vec<_>>();
             assert_eq!(events.len(), 2 + block.transactions.len());
 
             let mut transaction_leaves = vec![];

substrate/client/serai/tests/validator_sets.rs

@@ -3,6 +3,7 @@ use serai_abi::{
     address::SeraiAddress,
     network_id::{ExternalNetworkId, NetworkId},
     balance::Amount,
+    crypto::EmbeddedEllipticCurveKeys,
     validator_sets::{Session, ExternalValidatorSet, ValidatorSet, KeyShares},
   },
   validator_sets::Event,
@@ -44,67 +45,94 @@ async fn validator_sets() {
         panic!("finalized block remained the genesis block for over five minutes");
       };
 
+      // The genesis block should have the expected events
       {
         use sp_core::{Pair as _, sr25519::Pair};
         let genesis_validators = vec![(
           SeraiAddress::from(Pair::from_string("//Alice", None).unwrap().public()),
           KeyShares(1),
         )];
-        // The genesis block should have the expected events
+
         {
-          let mut events = serai
+          let events = serai
-            .as_of(serai.block_by_number(0).await.unwrap().unwrap().header.hash())
+            .events(serai.block_by_number(0).await.unwrap().unwrap().header.hash())
+            .await
+            .unwrap()
+            .validator_sets()
+            .set_embedded_elliptic_curve_keys_events()
+            .cloned()
+            .collect::<Vec<_>>();
+          assert_eq!(events.len(), ExternalNetworkId::all().collect::<Vec<_>>().len());
+
+          let state = serai.state().await.unwrap();
+          for (event, network) in events.into_iter().zip(ExternalNetworkId::all()) {
+            assert_eq!(
+              event,
+              Event::SetEmbeddedEllipticCurveKeys {
+                validator: genesis_validators[0].0,
+                keys: state
+                  .embedded_elliptic_curve_keys(genesis_validators[0].0, network)
+                  .await
+                  .unwrap()
+                  .unwrap(),
+              }
+            );
+          }
+        }
+
+        {
+          assert_eq!(
+            serai
+              .events(serai.block_by_number(0).await.unwrap().unwrap().header.hash())
               .await
               .unwrap()
               .validator_sets()
               .set_decided_events()
-            .await
+              .cloned()
-            .unwrap();
+              .collect::<Vec<_>>(),
-          events.sort_by_key(|event| borsh::to_vec(event).unwrap());
+            vec![
-          let mut expected = vec![
               Event::SetDecided {
                 set: ValidatorSet { network: NetworkId::Serai, session: Session(0) },
                 validators: genesis_validators.clone(),
               },
-            Event::SetDecided {
-              set: ValidatorSet { network: NetworkId::Serai, session: Session(1) },
-              validators: genesis_validators.clone(),
-            },
               Event::SetDecided {
                 set: ValidatorSet {
                   network: NetworkId::External(ExternalNetworkId::Bitcoin),
-                session: Session(0),
+                  session: Session(0)
                 },
                 validators: genesis_validators.clone(),
               },
               Event::SetDecided {
                 set: ValidatorSet {
                   network: NetworkId::External(ExternalNetworkId::Ethereum),
-                session: Session(0),
+                  session: Session(0)
                 },
                 validators: genesis_validators.clone(),
               },
               Event::SetDecided {
                 set: ValidatorSet {
                   network: NetworkId::External(ExternalNetworkId::Monero),
-                session: Session(0),
+                  session: Session(0)
                 },
                 validators: genesis_validators.clone(),
               },
-          ];
+              Event::SetDecided {
-          expected.sort_by_key(|event| borsh::to_vec(event).unwrap());
+                set: ValidatorSet { network: NetworkId::Serai, session: Session(1) },
-          assert_eq!(events, expected);
+                validators: genesis_validators.clone(),
+              },
+            ]
+          );
         }
 
         assert_eq!(
           serai
-            .as_of(serai.block_by_number(0).await.unwrap().unwrap().header.hash())
+            .events(serai.block_by_number(0).await.unwrap().unwrap().header.hash())
             .await
             .unwrap()
             .validator_sets()
             .accepted_handover_events()
-            .await
+            .cloned()
-            .unwrap(),
+            .collect::<Vec<_>>(),
           vec![Event::AcceptedHandover {
             set: ValidatorSet { network: NetworkId::Serai, session: Session(0) }
           }]
@@ -115,34 +143,30 @@ async fn validator_sets() {
       {
         assert_eq!(
           serai
-            .as_of(serai.block_by_number(1).await.unwrap().unwrap().header.hash())
+            .events(serai.block_by_number(1).await.unwrap().unwrap().header.hash())
             .await
             .unwrap()
             .validator_sets()
             .set_decided_events()
-            .await
+            .cloned()
-            .unwrap(),
+            .collect::<Vec<_>>(),
           vec![],
         );
         assert_eq!(
           serai
-            .as_of(serai.block_by_number(1).await.unwrap().unwrap().header.hash())
+            .events(serai.block_by_number(1).await.unwrap().unwrap().header.hash())
             .await
             .unwrap()
             .validator_sets()
             .accepted_handover_events()
-            .await
+            .cloned()
-            .unwrap(),
+            .collect::<Vec<_>>(),
           vec![],
         );
       }
 
       {
-        let serai = serai
+        let serai = serai.state().await.unwrap();
-          .as_of(serai.block_by_number(0).await.unwrap().unwrap().header.hash())
-          .await
-          .unwrap();
-        let serai = serai.validator_sets();
         for network in NetworkId::all() {
           match network {
             NetworkId::Serai => {

substrate/node/src/command.rs

@@ -57,25 +57,25 @@ pub fn run() -> sc_cli::Result<()> {
       cli.create_runner(cmd)?.sync_run(|config| cmd.run(config.chain_spec, config.network))
     }
 
-    Some(Subcommand::CheckBlock(cmd)) => cli.create_runner(cmd)?.async_run(|config| {
+    Some(Subcommand::CheckBlock(cmd)) => cli.create_runner(cmd)?.async_run(|mut config| {
       let PartialComponents { client, task_manager, import_queue, .. } =
-        service::new_partial(&config)?.0;
+        service::new_partial(&mut config)?.0;
       Ok((cmd.run(client, import_queue), task_manager))
     }),
 
-    Some(Subcommand::ExportBlocks(cmd)) => cli.create_runner(cmd)?.async_run(|config| {
+    Some(Subcommand::ExportBlocks(cmd)) => cli.create_runner(cmd)?.async_run(|mut config| {
-      let PartialComponents { client, task_manager, .. } = service::new_partial(&config)?.0;
+      let PartialComponents { client, task_manager, .. } = service::new_partial(&mut config)?.0;
       Ok((cmd.run(client, config.database), task_manager))
     }),
 
-    Some(Subcommand::ExportState(cmd)) => cli.create_runner(cmd)?.async_run(|config| {
+    Some(Subcommand::ExportState(cmd)) => cli.create_runner(cmd)?.async_run(|mut config| {
-      let PartialComponents { client, task_manager, .. } = service::new_partial(&config)?.0;
+      let PartialComponents { client, task_manager, .. } = service::new_partial(&mut config)?.0;
       Ok((cmd.run(client, config.chain_spec), task_manager))
     }),
 
-    Some(Subcommand::ImportBlocks(cmd)) => cli.create_runner(cmd)?.async_run(|config| {
+    Some(Subcommand::ImportBlocks(cmd)) => cli.create_runner(cmd)?.async_run(|mut config| {
       let PartialComponents { client, task_manager, import_queue, .. } =
-        service::new_partial(&config)?.0;
+        service::new_partial(&mut config)?.0;
       Ok((cmd.run(client, import_queue), task_manager))
     }),
 
@@ -83,9 +83,9 @@ pub fn run() -> sc_cli::Result<()> {
       cli.create_runner(cmd)?.sync_run(|config| cmd.run(config.database))
     }
 
-    Some(Subcommand::Revert(cmd)) => cli.create_runner(cmd)?.async_run(|config| {
+    Some(Subcommand::Revert(cmd)) => cli.create_runner(cmd)?.async_run(|mut config| {
       let PartialComponents { client, task_manager, backend, .. } =
-        service::new_partial(&config)?.0;
+        service::new_partial(&mut config)?.0;
       let aux_revert = Box::new(|client: Arc<FullClient>, backend, blocks| {
         sc_consensus_babe::revert(client.clone(), backend, blocks)?;
         sc_consensus_grandpa::revert(client, blocks)?;

substrate/node/src/service/mod.rs

@@ -65,7 +65,7 @@ fn create_inherent_data_providers(
 
 #[allow(clippy::type_complexity)]
 pub fn new_partial(
-  config: &Configuration,
+  config: &mut Configuration,
 ) -> Result<
   (
     PartialComponents<
@@ -79,6 +79,15 @@ pub fn new_partial(
   ),
   ServiceError,
 > {
+  // TODO: Copy events on block import, allow arbitrary pruning of state
+  config.state_pruning = Some(sc_service::PruningMode::ArchiveCanonical);
+  config.blocks_pruning = sc_service::BlocksPruning::KeepAll;
+
+  config.network.node_name = "serai".to_string();
+  config.network.client_version = "0.1.0".to_string();
+  config.network.listen_addresses =
+    vec!["/ip4/0.0.0.0/tcp/30333".parse().unwrap(), "/ip6/::/tcp/30333".parse().unwrap()];
+
   let telemetry = config
     .telemetry_endpoints
     .clone()
@@ -208,12 +217,7 @@ pub fn new_full(mut config: Configuration) -> Result<TaskManager, ServiceError>
       other: (block_import, babe_link, grandpa_link, shared_voter_state, mut telemetry),
     },
     keystore_container,
-  ) = new_partial(&config)?;
+  ) = new_partial(&mut config)?;
-
-  config.network.node_name = "serai".to_string();
-  config.network.client_version = "0.1.0".to_string();
-  config.network.listen_addresses =
-    vec!["/ip4/0.0.0.0/tcp/30333".parse().unwrap(), "/ip6/::/tcp/30333".parse().unwrap()];
 
   type N = sc_network::service::NetworkWorker<Block, <Block as sp_runtime::traits::Block>::Hash>;
   let mut net_config = sc_network::config::FullNetworkConfiguration::<_, _, N>::new(

substrate/primitives/src/network_id.rs

@@ -30,7 +30,6 @@ pub enum EmbeddedEllipticCurve {
   feature = "non_canonical_scale_derivations",
   derive(scale::Encode, scale::Decode, scale::MaxEncodedLen, scale::DecodeWithMemTracking)
 )]
-#[non_exhaustive]
 pub enum ExternalNetworkId {
   /// The Bitcoin network.
   Bitcoin = 1,

substrate/validator-sets/src/embedded_elliptic_curve_keys.rs

@@ -27,7 +27,7 @@ pub(crate) trait EmbeddedEllipticCurveKeys {
   fn set_embedded_elliptic_curve_keys(
     validator: Public,
     keys: SignedEmbeddedEllipticCurveKeys,
-  ) -> Result<(), ()>;
+  ) -> Result<EmbeddedEllipticCurveKeysStruct, ()>;
 
   /// Get a validator's embedded elliptic curve keys, for an external network.
   fn embedded_elliptic_curve_keys(
@@ -45,10 +45,10 @@ impl<S: EmbeddedEllipticCurveKeysStorage> EmbeddedEllipticCurveKeys for S {
   fn set_embedded_elliptic_curve_keys(
     validator: Public,
     keys: SignedEmbeddedEllipticCurveKeys,
-  ) -> Result<(), ()> {
+  ) -> Result<EmbeddedEllipticCurveKeysStruct, ()> {
     let keys = keys.verify(validator.into()).ok_or(())?;
-    S::EmbeddedEllipticCurveKeys::set(keys.network(), validator, Some(keys));
+    S::EmbeddedEllipticCurveKeys::insert(keys.network(), validator, keys);
-    Ok(())
+    Ok(keys)
   }
 
   /// Get a validator's embedded elliptic curve keys, for an external network.

substrate/validator-sets/src/lib.rs

@@ -193,10 +193,7 @@ mod pallet {
       for (participant, keys) in &self.participants {
         for (network, keys) in ExternalNetworkId::all().zip(keys.iter().cloned()) {
           assert_eq!(network, keys.network());
-          <Abstractions<T> as crate::EmbeddedEllipticCurveKeys>::set_embedded_elliptic_curve_keys(
+          Pallet::<T>::set_embedded_elliptic_curve_keys_internal(*participant, keys)
-            *participant,
-            keys,
-          )
             .expect("genesis embedded elliptic curve keys weren't valid");
         }
       }
@@ -331,6 +328,23 @@ mod pallet {
       )
     }
 
+    fn set_embedded_elliptic_curve_keys_internal(
+      validator: Public,
+      keys: SignedEmbeddedEllipticCurveKeys,
+    ) -> DispatchResult {
+      let network = keys.network();
+      let keys =
+        <Abstractions<T> as crate::EmbeddedEllipticCurveKeys>::set_embedded_elliptic_curve_keys(
+          validator, keys,
+        )
+        .map_err(|()| Error::<T>::InvalidEmbeddedEllipticCurveKeys)?;
+      Core::<T>::emit_event(Event::SetEmbeddedEllipticCurveKeys {
+        validator: validator.into(),
+        keys,
+      });
+      Ok(())
+    }
+
     /* TODO
     pub fn distribute_block_rewards(
       network: NetworkId,
@@ -492,16 +506,7 @@ mod pallet {
       keys: SignedEmbeddedEllipticCurveKeys,
     ) -> DispatchResult {
       let validator = ensure_signed(origin)?;
-      let network = keys.network();
+      Self::set_embedded_elliptic_curve_keys_internal(validator, keys)
-      <Abstractions<T> as crate::EmbeddedEllipticCurveKeys>::set_embedded_elliptic_curve_keys(
-        validator, keys,
-      )
-      .map_err(|()| Error::<T>::InvalidEmbeddedEllipticCurveKeys)?;
-      Core::<T>::emit_event(Event::SetEmbeddedEllipticCurveKeys {
-        validator: validator.into(),
-        network,
-      });
-      Ok(())
     }
 
     #[pallet::call_index(3)]