Continue filling out main loop

Adds generics to the db_channel macro, fixes the bug where it needed at least
one key.

common/db/src/create_db.rs

@@ -83,6 +83,18 @@ macro_rules! create_db {
           txn.del(&$field_name::key$(::<$($generic_name),+>)?($($arg),*));
           core::marker::PhantomData
         }
+
+        pub(crate) fn take$(<$($generic_name: $generic_type),+>)?(
+          txn: &mut impl DbTxn
+          $(, $arg: $arg_type)*
+        ) -> Option<$field_type> {
+          let key = $field_name::key$(::<$($generic_name),+>)?($($arg),*);
+          let res = txn.get(&key).map(|data| borsh::from_slice(data.as_ref()).unwrap());
+          if res.is_some() {
+            txn.del(key);
+          }
+          res
+        }
       }
     )*
   };
@@ -91,19 +103,30 @@ macro_rules! create_db {
 #[macro_export]
 macro_rules! db_channel {
   ($db_name: ident {
-    $($field_name: ident: ($($arg: ident: $arg_type: ty),*) -> $field_type: ty$(,)?)*
+    $($field_name: ident:
+      $(<$($generic_name: tt: $generic_type: tt),+>)?(
+        $($arg: ident: $arg_type: ty),*
+      ) -> $field_type: ty$(,)?
+    )*
   }) => {
     $(
       create_db! {
         $db_name {
-          $field_name: ($($arg: $arg_type,)* index: u32) -> $field_type,
+          $field_name: $(<$($generic_name: $generic_type),+>)?(
+            $($arg: $arg_type,)*
+            index: u32
+          ) -> $field_type
         }
       }
 
       impl $field_name {
-        pub(crate) fn send(txn: &mut impl DbTxn $(, $arg: $arg_type)*, value: &$field_type) {
+        pub(crate) fn send$(<$($generic_name: $generic_type),+>)?(
+          txn: &mut impl DbTxn
+          $(, $arg: $arg_type)*
+          , value: &$field_type
+        ) {
           // Use index 0 to store the amount of messages
-          let messages_sent_key = $field_name::key($($arg),*, 0);
+          let messages_sent_key = $field_name::key$(::<$($generic_name),+>)?($($arg,)* 0);
           let messages_sent = txn.get(&messages_sent_key).map(|counter| {
             u32::from_le_bytes(counter.try_into().unwrap())
           }).unwrap_or(0);
@@ -114,19 +137,22 @@ macro_rules! db_channel {
           // at the same time
           let index_to_use = messages_sent + 2;
 
-          $field_name::set(txn, $($arg),*, index_to_use, value);
+          $field_name::set$(::<$($generic_name),+>)?(txn, $($arg,)* index_to_use, value);
         }
-        pub(crate) fn try_recv(txn: &mut impl DbTxn $(, $arg: $arg_type)*) -> Option<$field_type> {
+        pub(crate) fn try_recv$(<$($generic_name: $generic_type),+>)?(
-          let messages_recvd_key = $field_name::key($($arg),*, 1);
+          txn: &mut impl DbTxn
+          $(, $arg: $arg_type)*
+        ) -> Option<$field_type> {
+          let messages_recvd_key = $field_name::key$(::<$($generic_name),+>)?($($arg,)* 1);
           let messages_recvd = txn.get(&messages_recvd_key).map(|counter| {
             u32::from_le_bytes(counter.try_into().unwrap())
           }).unwrap_or(0);
 
           let index_to_read = messages_recvd + 2;
 
-          let res = $field_name::get(txn, $($arg),*, index_to_read);
+          let res = $field_name::get$(::<$($generic_name),+>)?(txn, $($arg,)* index_to_read);
           if res.is_some() {
-            $field_name::del(txn, $($arg),*, index_to_read);
+            $field_name::del$(::<$($generic_name),+>)?(txn, $($arg,)* index_to_read);
             txn.put(&messages_recvd_key, (messages_recvd + 1).to_le_bytes());
           }
           res

processor/bitcoin/src/db.rs

@@ -1,10 +1,19 @@
+use ciphersuite::group::GroupEncoding;
+
 use serai_client::validator_sets::primitives::Session;
 
-use serai_db::{Get, DbTxn, create_db};
+use serai_db::{Get, DbTxn, create_db, db_channel};
+use primitives::EncodableG;
 
 create_db! {
   Processor {
-    ExternalKeyForSession: (session: Session) -> Vec<u8>,
+    ExternalKeyForSessionForSigners: <K: GroupEncoding>(session: Session) -> EncodableG<K>,
+  }
+}
+
+db_channel! {
+  Processor {
+    KeyToActivate: <K: GroupEncoding>() -> EncodableG<K>
   }
 }
 

processor/bitcoin/src/key_gen.rs

@@ -1,7 +1,7 @@
 use ciphersuite::{group::GroupEncoding, Ciphersuite, Secp256k1};
 use frost::ThresholdKeys;
 
-use crate::scan::scanner;
+use crate::{primitives::x_coord_to_even_point, scan::scanner};
 
 pub(crate) struct KeyGenParams;
 impl key_gen::KeyGenParams for KeyGenParams {
@@ -21,4 +21,8 @@ impl key_gen::KeyGenParams for KeyGenParams {
     // Skip the parity encoding as we know this key is even
     key[1 ..].to_vec()
   }
+
+  fn decode_key(key: &[u8]) -> Option<<Self::ExternalNetworkCurve as Ciphersuite>::G> {
+    x_coord_to_even_point(key)
+  }
 }

processor/bitcoin/src/main.rs

@@ -9,9 +9,11 @@ static ALLOCATOR: zalloc::ZeroizingAlloc<std::alloc::System> =
 use ciphersuite::Ciphersuite;
 
 use serai_db::{DbTxn, Db};
+use ::primitives::EncodableG;
+use ::key_gen::KeyGenParams as KeyGenParamsTrait;
 
 mod primitives;
-pub(crate) use primitives::*;
+pub(crate) use crate::primitives::*;
 
 // Internal utilities for scanning transactions
 mod scan;
@@ -50,59 +52,123 @@ async fn send_message(_msg: messages::ProcessorMessage) {
 
 async fn coordinator_loop<D: Db>(
   mut db: D,
-  mut key_gen: ::key_gen::KeyGen<KeyGenParams, D>,
+  mut key_gen: ::key_gen::KeyGen<KeyGenParams>,
   mut signers: signers::Signers<D, Rpc<D>, Scheduler<D>, Rpc<D>>,
   mut scanner: Option<scanner::Scanner<Rpc<D>>>,
 ) {
   loop {
-    let mut txn = Some(db.txn());
+    let mut txn = db.txn();
-    let msg = next_message(txn.as_mut().unwrap()).await;
+    let msg = next_message(&mut txn).await;
+    let mut txn = Some(txn);
     match msg {
       messages::CoordinatorMessage::KeyGen(msg) => {
+        let txn = txn.as_mut().unwrap();
+        let mut new_key = None;
         // This is a computationally expensive call yet it happens infrequently
-        for msg in key_gen.handle(txn.as_mut().unwrap(), msg) {
+        for msg in key_gen.handle(txn, msg) {
+          if let messages::key_gen::ProcessorMessage::GeneratedKeyPair { session, .. } = &msg {
+            new_key = Some(*session)
+          }
           send_message(messages::ProcessorMessage::KeyGen(msg)).await;
         }
+
+        // If we were yielded a key, register it in the signers
+        if let Some(session) = new_key {
+          let (substrate_keys, network_keys) =
+            ::key_gen::KeyGen::<KeyGenParams>::key_shares(txn, session)
+              .expect("generated key pair yet couldn't get key shares");
+          signers.register_keys(txn, session, substrate_keys, network_keys);
         }
+      }
+
       // These are cheap calls which are fine to be here in this loop
-      messages::CoordinatorMessage::Sign(msg) => signers.queue_message(txn.as_mut().unwrap(), &msg),
+      messages::CoordinatorMessage::Sign(msg) => {
+        let txn = txn.as_mut().unwrap();
+        signers.queue_message(txn, &msg)
+      }
       messages::CoordinatorMessage::Coordinator(
         messages::coordinator::CoordinatorMessage::CosignSubstrateBlock {
           session,
           block_number,
           block,
         },
-      ) => signers.cosign_block(txn.take().unwrap(), session, block_number, block),
+      ) => {
+        let txn = txn.take().unwrap();
+        signers.cosign_block(txn, session, block_number, block)
+      }
       messages::CoordinatorMessage::Coordinator(
         messages::coordinator::CoordinatorMessage::SignSlashReport { session, report },
-      ) => signers.sign_slash_report(txn.take().unwrap(), session, &report),
+      ) => {
+        let txn = txn.take().unwrap();
+        signers.sign_slash_report(txn, session, &report)
+      }
+
       messages::CoordinatorMessage::Substrate(msg) => match msg {
         messages::substrate::CoordinatorMessage::SetKeys { serai_time, session, key_pair } => {
-          db::ExternalKeyForSession::set(txn.as_mut().unwrap(), session, &key_pair.1.into_inner());
+          let txn = txn.as_mut().unwrap();
-          todo!("TODO: Register in signers");
+          let key = EncodableG(
-          todo!("TODO: Scanner activation")
+            KeyGenParams::decode_key(key_pair.1.as_ref()).expect("invalid key set on serai"),
+          );
+
+          // Queue the key to be activated upon the next Batch
+          db::KeyToActivate::send::<
+            <<KeyGenParams as ::key_gen::KeyGenParams>::ExternalNetworkCurve as Ciphersuite>::G,
+          >(txn, &key);
+
+          // Set the external key, as needed by the signers
+          db::ExternalKeyForSessionForSigners::set::<
+            <<KeyGenParams as ::key_gen::KeyGenParams>::ExternalNetworkCurve as Ciphersuite>::G,
+          >(txn, session, &key);
+
+          // This isn't cheap yet only happens for the very first set of keys
+          if scanner.is_none() {
+            todo!("TODO")
+          }
         }
         messages::substrate::CoordinatorMessage::SlashesReported { session } => {
-          let key_bytes = db::ExternalKeyForSession::get(txn.as_ref().unwrap(), session).unwrap();
+          let txn = txn.as_mut().unwrap();
-          let mut key_bytes = key_bytes.as_slice();
-          let key =
-            <KeyGenParams as ::key_gen::KeyGenParams>::ExternalNetworkCurve::read_G(&mut key_bytes)
-              .unwrap();
-          assert!(key_bytes.is_empty());
 
-          signers.retire_session(txn.as_mut().unwrap(), session, &key)
+          // Since this session had its slashes reported, it has finished all its signature
+          // protocols and has been fully retired. We retire it from the signers accordingly
+          let key = db::ExternalKeyForSessionForSigners::take::<
+            <<KeyGenParams as ::key_gen::KeyGenParams>::ExternalNetworkCurve as Ciphersuite>::G,
+          >(txn, session)
+          .unwrap()
+          .0;
+
+          // This is a cheap call
+          signers.retire_session(txn, session, &key)
         }
         messages::substrate::CoordinatorMessage::BlockWithBatchAcknowledgement {
-          block,
+          block: _,
+          batch_id,
+          in_instruction_succeededs,
+          burns,
+        } => {
+          let mut txn = txn.take().unwrap();
+          let scanner = scanner.as_mut().unwrap();
+          let key_to_activate = db::KeyToActivate::try_recv::<
+            <<KeyGenParams as ::key_gen::KeyGenParams>::ExternalNetworkCurve as Ciphersuite>::G,
+          >(&mut txn)
+          .map(|key| key.0);
+          // This is a cheap call as it internally just queues this to be done later
+          scanner.acknowledge_batch(
+            txn,
             batch_id,
             in_instruction_succeededs,
             burns,
             key_to_activate,
-        } => todo!("TODO"),
+          )
+        }
         messages::substrate::CoordinatorMessage::BlockWithoutBatchAcknowledgement {
-          block,
+          block: _,
           burns,
-        } => todo!("TODO"),
+        } => {
+          let txn = txn.take().unwrap();
+          let scanner = scanner.as_mut().unwrap();
+          // This is a cheap call as it internally just queues this to be done later
+          scanner.queue_burns(txn, burns)
+        }
       },
     };
     // If the txn wasn't already consumed and committed, commit it

processor/bitcoin/src/primitives/mod.rs

@@ -1,3 +1,20 @@
+use ciphersuite::{Ciphersuite, Secp256k1};
+
+use bitcoin_serai::bitcoin::key::{Parity, XOnlyPublicKey};
+
 pub(crate) mod output;
 pub(crate) mod transaction;
 pub(crate) mod block;
+
+pub(crate) fn x_coord_to_even_point(key: &[u8]) -> Option<<Secp256k1 as Ciphersuite>::G> {
+  if key.len() != 32 {
+    None?
+  };
+
+  // Read the x-only public key
+  let key = XOnlyPublicKey::from_slice(key).ok()?;
+  // Convert to a full public key
+  let key = key.public_key(Parity::Even);
+  // Convert to k256 (from libsecp256k1)
+  Secp256k1::read_G(&mut key.serialize().as_slice()).ok()
+}

processor/bitcoin/src/primitives/output.rs

@@ -4,11 +4,7 @@ use ciphersuite::{Ciphersuite, Secp256k1};
 
 use bitcoin_serai::{
   bitcoin::{
-    hashes::Hash as HashTrait,
+    hashes::Hash as HashTrait, consensus::Encodable, script::Instruction, transaction::Transaction,
-    key::{Parity, XOnlyPublicKey},
-    consensus::Encodable,
-    script::Instruction,
-    transaction::Transaction,
   },
   wallet::ReceivedOutput as WalletOutput,
 };
@@ -24,7 +20,10 @@ use serai_client::{
 
 use primitives::{OutputType, ReceivedOutput};
 
-use crate::scan::{offsets_for_key, presumed_origin, extract_serai_data};
+use crate::{
+  primitives::x_coord_to_even_point,
+  scan::{offsets_for_key, presumed_origin, extract_serai_data},
+};
 
 #[derive(Clone, PartialEq, Eq, Hash, Debug, Encode, Decode, BorshSerialize, BorshDeserialize)]
 pub(crate) struct OutputId([u8; 36]);
@@ -117,15 +116,11 @@ impl ReceivedOutput<<Secp256k1 as Ciphersuite>::G, Address> for Output {
     let Instruction::PushBytes(key) = script.instructions_minimal().last().unwrap().unwrap() else {
       panic!("last item in v1 Taproot script wasn't bytes")
     };
-    let key = XOnlyPublicKey::from_slice(key.as_ref())
+    let key = x_coord_to_even_point(key.as_ref())
-      .expect("last item in v1 Taproot script wasn't a valid x-only public key");
+      .expect("last item in scanned v1 Taproot script wasn't a valid x-only public key");
 
-    // Convert to a full key
-    let key = key.public_key(Parity::Even);
-    // Convert to a k256 key (from libsecp256k1)
-    let output_key = Secp256k1::read_G(&mut key.serialize().as_slice()).unwrap();
     // The output's key minus the output's offset is the root key
-    output_key - (<Secp256k1 as Ciphersuite>::G::GENERATOR * self.output.offset())
+    key - (<Secp256k1 as Ciphersuite>::G::GENERATOR * self.output.offset())
   }
 
   fn presumed_origin(&self) -> Option<Address> {

processor/key-gen/src/lib.rs

@@ -20,7 +20,7 @@ use dkg::{Participant, ThresholdKeys, evrf::*};
 use serai_validator_sets_primitives::Session;
 use messages::key_gen::*;
 
-use serai_db::{DbTxn, Db};
+use serai_db::{Get, DbTxn};
 
 mod generators;
 use generators::generators;
@@ -49,6 +49,17 @@ pub trait KeyGenParams {
   fn encode_key(key: <Self::ExternalNetworkCurve as Ciphersuite>::G) -> Vec<u8> {
     key.to_bytes().as_ref().to_vec()
   }
+
+  /// Decode keys from their optimal encoding.
+  ///
+  /// A default implementation is provided which calls the traditional `from_bytes`.
+  fn decode_key(mut key: &[u8]) -> Option<<Self::ExternalNetworkCurve as Ciphersuite>::G> {
+    let res = <Self::ExternalNetworkCurve as Ciphersuite>::read_G(&mut key).ok()?;
+    if !key.is_empty() {
+      None?;
+    }
+    Some(res)
+  }
 }
 
 /*
@@ -128,47 +139,41 @@ fn coerce_keys<C: EvrfCurve>(
 
 /// An instance of the Serai key generation protocol.
 #[derive(Debug)]
-pub struct KeyGen<P: KeyGenParams, D: Db> {
+pub struct KeyGen<P: KeyGenParams> {
-  db: D,
   substrate_evrf_private_key:
     Zeroizing<<<Ristretto as EvrfCurve>::EmbeddedCurve as Ciphersuite>::F>,
   network_evrf_private_key:
     Zeroizing<<<P::ExternalNetworkCurve as EvrfCurve>::EmbeddedCurve as Ciphersuite>::F>,
 }
 
-impl<P: KeyGenParams, D: Db> KeyGen<P, D> {
+impl<P: KeyGenParams> KeyGen<P> {
   /// Create a new key generation instance.
   #[allow(clippy::new_ret_no_self)]
   pub fn new(
-    db: D,
     substrate_evrf_private_key: Zeroizing<
       <<Ristretto as EvrfCurve>::EmbeddedCurve as Ciphersuite>::F,
     >,
     network_evrf_private_key: Zeroizing<
       <<P::ExternalNetworkCurve as EvrfCurve>::EmbeddedCurve as Ciphersuite>::F,
     >,
-  ) -> KeyGen<P, D> {
+  ) -> KeyGen<P> {
-    KeyGen { db, substrate_evrf_private_key, network_evrf_private_key }
+    KeyGen { substrate_evrf_private_key, network_evrf_private_key }
   }
 
   /// Fetch the key shares for a specific session.
   #[allow(clippy::type_complexity)]
   pub fn key_shares(
-    &self,
+    getter: &impl Get,
     session: Session,
   ) -> Option<(Vec<ThresholdKeys<Ristretto>>, Vec<ThresholdKeys<P::ExternalNetworkCurve>>)> {
     // This is safe, despite not having a txn, since it's a static value
     // It doesn't change over time/in relation to other operations
     // It is solely set or unset
-    KeyGenDb::<P>::key_shares(&self.db, session)
+    KeyGenDb::<P>::key_shares(getter, session)
   }
 
   /// Handle a message from the coordinator.
-  pub fn handle(
+  pub fn handle(&mut self, txn: &mut impl DbTxn, msg: CoordinatorMessage) -> Vec<ProcessorMessage> {
-    &mut self,
-    txn: &mut D::Transaction<'_>,
-    msg: CoordinatorMessage,
-  ) -> Vec<ProcessorMessage> {
     const SUBSTRATE_KEY_CONTEXT: &[u8] = b"substrate";
     const NETWORK_KEY_CONTEXT: &[u8] = b"network";
     fn context<P: KeyGenParams>(session: Session, key_context: &[u8]) -> [u8; 32] {
@@ -292,7 +297,7 @@ impl<P: KeyGenParams, D: Db> KeyGen<P, D> {
             // If we've already generated these keys, we don't actually need to save these
             // participations and continue. We solely have to verify them, as to identify malicious
             // participants and prevent DoSs, before returning
-            if self.key_shares(session).is_some() {
+            if Self::key_shares(txn, session).is_some() {
               log::debug!("already finished generating a key for {:?}", session);
 
               match EvrfDkg::<Ristretto>::verify(