Delay cosign acknowledgments

We don't have consensus on the session's last block, so we shouldn't check if
the cosign is before the session ends. What matters is that network, within its
set, claims it's still active at that block (on its view of the blockchain).

Cargo.lock

@@ -3547,7 +3547,7 @@ dependencies = [
  "httpdate",
  "itoa",
  "pin-project-lite",
- "socket2 0.4.10",
+ "socket2 0.5.8",
  "tokio",
  "tower-service",
  "tracing",
@@ -4111,7 +4111,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fc2f4eb4bc735547cfed7c0a4922cbd04a4655978c09b54f1f7b228750664c34"
 dependencies = [
  "cfg-if",
- "windows-targets 0.52.6",
+ "windows-targets 0.48.5",
 ]
 
 [[package]]
@@ -4601,14 +4601,13 @@ dependencies = [
 
 [[package]]
 name = "librocksdb-sys"
-version = "0.16.0+8.10.0"
+version = "0.17.1+9.9.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ce3d60bc059831dc1c83903fb45c103f75db65c5a7bf22272764d9cc683e348c"
+checksum = "2b7869a512ae9982f4d46ba482c2a304f1efd80c6412a3d4bf57bb79a619679f"
 dependencies = [
  "bindgen",
  "bzip2-sys",
  "cc",
- "glob",
  "libc",
  "libz-sys",
  "lz4-sys",
@@ -6764,14 +6763,14 @@ dependencies = [
 name = "rocksdb"
 version = "0.21.0"
 dependencies = [
- "rocksdb 0.22.0",
+ "rocksdb 0.23.0",
 ]
 
 [[package]]
 name = "rocksdb"
-version = "0.22.0"
+version = "0.23.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6bd13e55d6d7b8cd0ea569161127567cd587676c99f4472f779a0279aa60a7a7"
+checksum = "26ec73b20525cb235bad420f911473b69f9fe27cc856c5461bccd7e4af037f43"
 dependencies = [
  "libc",
  "librocksdb-sys",
@@ -8371,14 +8370,12 @@ version = "0.1.0"
 dependencies = [
  "blake2",
  "borsh",
- "ciphersuite",
  "log",
  "parity-scale-codec",
- "schnorr-signatures",
+ "schnorrkel",
  "serai-client",
  "serai-db",
  "serai-task",
- "sp-application-crypto",
  "tokio",
 ]
 
@@ -11696,7 +11693,7 @@ version = "0.1.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cf221c93e13a30d793f7645a0e7762c55d169dbb0a49671918a2319d289b10bb"
 dependencies = [
- "windows-sys 0.48.0",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]

common/db/src/create_db.rs

@@ -143,6 +143,19 @@ macro_rules! db_channel {
 
           Self::set(txn, $($arg,)* index_to_use, value);
         }
+        pub(crate) fn peek(
+          getter: &impl Get
+          $(, $arg: $arg_type)*
+        ) -> Option<$field_type> {
+          let messages_recvd_key = Self::key($($arg,)* 1);
+          let messages_recvd = getter.get(&messages_recvd_key).map(|counter| {
+            u32::from_le_bytes(counter.try_into().unwrap())
+          }).unwrap_or(0);
+
+          let index_to_read = messages_recvd + 2;
+
+          Self::get(getter, $($arg,)* index_to_read)
+        }
         pub(crate) fn try_recv(
           txn: &mut impl DbTxn
           $(, $arg: $arg_type)*

coordinator/cosign/README.md

@@ -66,7 +66,7 @@ to exhibit the same behavior), yet prevents interaction with it.
 
 If cosigns representing 17% of the non-Serai validators sets by weight are
 detected for distinct blocks at the same position, the protocol halts. An
-explicit latency period of five seconds is enacted after receiving a cosign
+explicit latency period of seventy seconds is enacted after receiving a cosign
 commit for the detection of such an equivocation. This is largely redundant
 given how the Serai blockchain node will presumably have halted itself by this
 time.
@@ -114,8 +114,8 @@ asynchronous network or 11.33% of non-Serai validator sets' stake.
 ### TODO
 
 The Serai node no longer responding to RPC requests upon detecting any
-equivocation, the delayed acknowledgement of cosigns, and the fallback protocol
-where validators individually produce signatures, are not implemented at this
-time. The former means the detection of equivocating cosigns not redundant and
-the latter makes 5.67% of non-Serai validator sets' stake the DoS threshold,
-even without control of an asynchronous network.
+equivocation, and the fallback protocol where validators individually produce
+signatures, are not implemented at this time. The former means the detection of
+equivocating cosigns is not redundant and the latter makes 5.67% of non-Serai
+validator sets' stake the DoS threshold, even without control of an
+asynchronous network.

coordinator/cosign/src/delay.rs

@@ -0,0 +1,55 @@
+use core::future::Future;
+use std::time::{Duration, SystemTime};
+
+use serai_db::*;
+use serai_task::ContinuallyRan;
+
+use crate::evaluator::CosignedBlocks;
+
+/// How often callers should broadcast the cosigns flagged for rebroadcasting.
+pub const BROADCAST_FREQUENCY: Duration = Duration::from_secs(60);
+const SYNCHRONY_EXPECTATION: Duration = Duration::from_secs(10);
+const ACKNOWLEDGEMENT_DELAY: Duration =
+  Duration::from_secs(BROADCAST_FREQUENCY.as_secs() + SYNCHRONY_EXPECTATION.as_secs());
+
+create_db!(
+  SubstrateCosignDelay {
+    // The latest cosigned block number.
+    LatestCosignedBlockNumber: () -> u64,
+  }
+);
+
+/// A task to delay acknowledgement of cosigns.
+pub(crate) struct CosignDelayTask<D: Db> {
+  pub(crate) db: D,
+}
+
+impl<D: Db> ContinuallyRan for CosignDelayTask<D> {
+  fn run_iteration(&mut self) -> impl Send + Future<Output = Result<bool, String>> {
+    async move {
+      let mut made_progress = false;
+      loop {
+        let mut txn = self.db.txn();
+
+        // Receive the next block to mark as cosigned
+        let Some((block_number, time_evaluated)) = CosignedBlocks::try_recv(&mut txn) else {
+          break;
+        };
+        // Calculate when we should mark it as valid
+        let time_valid =
+          SystemTime::UNIX_EPOCH + Duration::from_secs(time_evaluated) + ACKNOWLEDGEMENT_DELAY;
+        // Sleep until then
+        tokio::time::sleep(SystemTime::now().duration_since(time_valid).unwrap_or(Duration::ZERO))
+          .await;
+
+        // Set the cosigned block
+        LatestCosignedBlockNumber::set(&mut txn, &block_number);
+        txn.commit();
+
+        made_progress = true;
+      }
+
+      Ok(made_progress)
+    }
+  }
+}

coordinator/cosign/src/evaluator.rs

@@ -1,111 +1,115 @@
 use core::future::Future;
-
-use serai_client::{primitives::Amount, Serai};
+use std::time::{Duration, SystemTime};
 
 use serai_db::*;
 use serai_task::ContinuallyRan;
 
 use crate::{
-  *,
-  intend::{BlockEventData, BlockEvents},
+  HasEvents, GlobalSession, NetworksLatestCosignedBlock, RequestNotableCosigns,
+  intend::{GlobalSessionsChannel, BlockEventData, BlockEvents},
 };
 
 create_db!(
   SubstrateCosignEvaluator {
-    // The latest cosigned block number.
-    LatestCosignedBlockNumber: () -> u64,
-    // The latest global session evaluated.
-    // TODO: Also include the weights here
-    LatestGlobalSessionEvaluated: () -> ([u8; 32], Vec<ValidatorSet>),
+    // The global session currently being evaluated.
+    CurrentlyEvaluatedGlobalSession: () -> ([u8; 32], GlobalSession),
   }
 );
 
+db_channel!(
+  SubstrateCosignEvaluatorChannels {
+    // (cosigned block, time cosign was evaluated)
+    CosignedBlocks: () -> (u64, u64),
+  }
+);
+
+// This is a strict function which won't panic, even with a malicious Serai node, so long as:
+// - It's called incrementally
+// - It's only called for block numbers we've completed indexing on within the intend task
+// - It's only called for block numbers after a global session has started
+// - The global sessions channel is populated as the block declaring the session is indexed
+// Which all hold true within the context of this task and the intend task.
+//
+// This function will also ensure the currently evaluated global session is incremented once we
+// finish evaluation of the prior session.
+fn currently_evaluated_global_session_strict(
+  txn: &mut impl DbTxn,
+  block_number: u64,
+) -> ([u8; 32], GlobalSession) {
+  let mut res = {
+    let existing = match CurrentlyEvaluatedGlobalSession::get(txn) {
+      Some(existing) => existing,
+      None => {
+        let first = GlobalSessionsChannel::try_recv(txn)
+          .expect("fetching latest global session yet none declared");
+        CurrentlyEvaluatedGlobalSession::set(txn, &first);
+        first
+      }
+    };
+    assert!(
+      existing.1.start_block_number <= block_number,
+      "candidate's start block number exceeds our block number"
+    );
+    existing
+  };
+
+  if let Some(next) = GlobalSessionsChannel::peek(txn) {
+    assert!(
+      block_number <= next.1.start_block_number,
+      "currently_evaluated_global_session_strict wasn't called incrementally"
+    );
+    // If it's time for this session to activate, take it from the channel and set it
+    if block_number == next.1.start_block_number {
+      GlobalSessionsChannel::try_recv(txn).unwrap();
+      CurrentlyEvaluatedGlobalSession::set(txn, &next);
+      res = next;
+    }
+  }
+
+  res
+}
+
 /// A task to determine if a block has been cosigned and we should handle it.
 pub(crate) struct CosignEvaluatorTask<D: Db, R: RequestNotableCosigns> {
   pub(crate) db: D,
-  pub(crate) serai: Serai,
   pub(crate) request: R,
 }
 
-async fn get_latest_global_session_evaluated(
-  txn: &mut impl DbTxn,
-  serai: &Serai,
-  parent_hash: [u8; 32],
-) -> Result<([u8; 32], Vec<ValidatorSet>), String> {
-  Ok(match LatestGlobalSessionEvaluated::get(txn) {
-    Some(res) => res,
-    None => {
-      // This is the initial global session
-      // Fetch the sets participating and declare it the latest value recognized
-      let sets = cosigning_sets_by_parent_hash(serai, parent_hash).await?;
-      let initial_global_session = GlobalSession::new(sets.clone()).id();
-      LatestGlobalSessionEvaluated::set(txn, &(initial_global_session, sets.clone()));
-      (initial_global_session, sets)
-    }
-  })
-}
-
 impl<D: Db, R: RequestNotableCosigns> ContinuallyRan for CosignEvaluatorTask<D, R> {
   fn run_iteration(&mut self) -> impl Send + Future<Output = Result<bool, String>> {
     async move {
-      let latest_cosigned_block_number = LatestCosignedBlockNumber::get(&self.db).unwrap_or(0);
-
       let mut known_cosign = None;
       let mut made_progress = false;
       loop {
         let mut txn = self.db.txn();
-        let Some(BlockEventData { block_number, parent_hash, block_hash, has_events }) =
-          BlockEvents::try_recv(&mut txn)
+        let Some(BlockEventData { block_number, has_events }) = BlockEvents::try_recv(&mut txn)
         else {
           break;
         };
-        // Make sure these two feeds haven't desynchronized somehow
-        // We could remove our `LatestCosignedBlockNumber`, making the latest cosigned block number
-        // the next message in the channel's block number minus one, but that'd only work when the
-        // channel isn't empty
-        assert_eq!(block_number, latest_cosigned_block_number + 1);
 
         match has_events {
           // Because this had notable events, we require an explicit cosign for this block by a
           // supermajority of the prior block's validator sets
           HasEvents::Notable => {
-            let (global_session, sets) =
-              get_latest_global_session_evaluated(&mut txn, &self.serai, parent_hash).await?;
+            let (global_session, global_session_info) =
+              currently_evaluated_global_session_strict(&mut txn, block_number);
 
             let mut weight_cosigned = 0;
-            let mut total_weight = 0;
-            let (_, global_session_start_block) = GlobalSessions::get(&txn, global_session)
-              .ok_or_else(|| {
-                "checking if intended cosign was satisfied within an unrecognized global session"
-                  .to_string()
-              })?;
-            for set in sets {
-              // Fetch the weight for this set, as of the start of the global session
-              // This simplifies the logic around which set of stakes to use when evaluating
-              // cosigns, even if it's lossy as it isn't accurate to how stake may fluctuate within
-              // a session
-              let stake = self
-                .serai
-                .as_of(global_session_start_block)
-                .validator_sets()
-                .total_allocated_stake(set.network)
-                .await
-                .map_err(|e| format!("{e:?}"))?
-                .unwrap_or(Amount(0))
-                .0;
-              total_weight += stake;
-
+            for set in global_session_info.sets {
               // Check if we have the cosign from this set
               if NetworksLatestCosignedBlock::get(&txn, global_session, set.network)
                 .map(|signed_cosign| signed_cosign.cosign.block_number) ==
                 Some(block_number)
               {
                 // Since have this cosign, add the set's weight to the weight which has cosigned
-                weight_cosigned += stake;
+                weight_cosigned +=
+                  global_session_info.stakes.get(&set.network).ok_or_else(|| {
+                    "ValidatorSet in global session yet didn't have its stake".to_string()
+                  })?;
               }
             }
             // Check if the sum weight doesn't cross the required threshold
-            if weight_cosigned < (((total_weight * 83) / 100) + 1) {
+            if weight_cosigned < (((global_session_info.total_stake * 83) / 100) + 1) {
               // Request the necessary cosigns over the network
               // TODO: Add a timer to ensure this isn't called too often
               self
@@ -118,13 +122,6 @@ impl<D: Db, R: RequestNotableCosigns> ContinuallyRan for CosignEvaluatorTask<D,
                 "notable block (#{block_number}) wasn't yet cosigned. this should resolve shortly",
               ));
             }
-
-            // Since this block changes the global session, update it
-            {
-              let sets = cosigning_sets(&self.serai.as_of(block_hash)).await?;
-              let global_session = GlobalSession::new(sets.clone()).id();
-              LatestGlobalSessionEvaluated::set(&mut txn, &(global_session, sets));
-            }
           }
           // Since this block didn't have any notable events, we simply require a cosign for this
           // block or a greater block by the current validator sets
@@ -147,30 +144,12 @@ impl<D: Db, R: RequestNotableCosigns> ContinuallyRan for CosignEvaluatorTask<D,
               */
 
               // Get the global session for this block
-              let (global_session, sets) =
-                get_latest_global_session_evaluated(&mut txn, &self.serai, parent_hash).await?;
-              let (_, global_session_start_block) = GlobalSessions::get(&txn, global_session)
-                .ok_or_else(|| {
-                  "checking if intended cosign was satisfied within an unrecognized global session"
-                    .to_string()
-                })?;
+              let (global_session, global_session_info) =
+                currently_evaluated_global_session_strict(&mut txn, block_number);
 
               let mut weight_cosigned = 0;
-              let mut total_weight = 0;
               let mut lowest_common_block: Option<u64> = None;
-              for set in sets {
-                let stake = self
-                  .serai
-                  .as_of(global_session_start_block)
-                  .validator_sets()
-                  .total_allocated_stake(set.network)
-                  .await
-                  .map_err(|e| format!("{e:?}"))?
-                  .unwrap_or(Amount(0))
-                  .0;
-                // Increment total_weight with this set's stake
-                total_weight += stake;
-
+              for set in global_session_info.sets {
                 // Check if this set cosigned this block or not
                 let Some(cosign) =
                   NetworksLatestCosignedBlock::get(&txn, global_session, set.network)
@@ -178,7 +157,10 @@ impl<D: Db, R: RequestNotableCosigns> ContinuallyRan for CosignEvaluatorTask<D,
                   continue;
                 };
                 if cosign.cosign.block_number >= block_number {
-                  weight_cosigned += total_weight
+                  weight_cosigned +=
+                    global_session_info.stakes.get(&set.network).ok_or_else(|| {
+                      "ValidatorSet in global session yet didn't have its stake".to_string()
+                    })?;
                 }
 
                 // Update the lowest block common to all of these cosigns
@@ -188,7 +170,7 @@ impl<D: Db, R: RequestNotableCosigns> ContinuallyRan for CosignEvaluatorTask<D,
               }
 
               // Check if the sum weight doesn't cross the required threshold
-              if weight_cosigned < (((total_weight * 83) / 100) + 1) {
+              if weight_cosigned < (((global_session_info.total_stake * 83) / 100) + 1) {
                 // Request the superseding notable cosigns over the network
                 // If this session hasn't yet produced notable cosigns, then we presume we'll see
                 // the desired non-notable cosigns as part of normal operations, without needing to
@@ -218,8 +200,17 @@ impl<D: Db, R: RequestNotableCosigns> ContinuallyRan for CosignEvaluatorTask<D,
           HasEvents::No => {}
         }
 
-        // Since we checked we had the necessary cosigns, increment the latest cosigned block
-        LatestCosignedBlockNumber::set(&mut txn, &block_number);
+        // Since we checked we had the necessary cosigns, send it for delay before acknowledgement
+        CosignedBlocks::send(
+          &mut txn,
+          &(
+            block_number,
+            SystemTime::now()
+              .duration_since(SystemTime::UNIX_EPOCH)
+              .unwrap_or(Duration::ZERO)
+              .as_secs(),
+          ),
+        );
         txn.commit();
 
         made_progress = true;

coordinator/cosign/src/intend.rs

@@ -1,6 +1,11 @@
 use core::future::Future;
+use std::collections::HashMap;
 
-use serai_client::{Serai, validator_sets::primitives::ValidatorSet};
+use serai_client::{
+  primitives::{SeraiAddress, Amount},
+  validator_sets::primitives::ValidatorSet,
+  Serai,
+};
 
 use serai_db::*;
 use serai_task::ContinuallyRan;
@@ -16,13 +21,12 @@ create_db!(
 #[derive(Debug, BorshSerialize, BorshDeserialize)]
 pub(crate) struct BlockEventData {
   pub(crate) block_number: u64,
-  pub(crate) parent_hash: [u8; 32],
-  pub(crate) block_hash: [u8; 32],
   pub(crate) has_events: HasEvents,
 }
 
 db_channel! {
   CosignIntendChannels {
+    GlobalSessionsChannel: () -> ([u8; 32], GlobalSession),
     BlockEvents: () -> BlockEventData,
     IntendedCosigns: (set: ValidatorSet) -> CosignIntent,
   }
@@ -71,57 +75,101 @@ impl<D: Db> ContinuallyRan for CosignIntendTask<D> {
             .await
             .map_err(|e| format!("{e:?}"))?;
 
+        // Check we are indexing a linear chain
+        if (block_number > 1) &&
+          (<[u8; 32]>::from(block.header.parent_hash) !=
+            SubstrateBlocks::get(&txn, block_number - 1)
+              .expect("indexing a block but haven't indexed its parent"))
+        {
+          Err(format!(
+            "node's block #{block_number} doesn't build upon the block #{} prior indexed",
+            block_number - 1
+          ))?;
+        }
+        SubstrateBlocks::set(&mut txn, block_number, &block.hash());
+
+        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());
+          let sets_and_keys = cosigning_sets(&serai).await?;
+          let global_session =
+            GlobalSession::id(sets_and_keys.iter().map(|(set, _key)| *set).collect());
+
+          let mut sets = Vec::with_capacity(sets_and_keys.len());
+          let mut keys = HashMap::with_capacity(sets_and_keys.len());
+          let mut stakes = HashMap::with_capacity(sets_and_keys.len());
+          let mut total_stake = 0;
+          for (set, key) in &sets_and_keys {
+            sets.push(*set);
+            keys.insert(set.network, SeraiAddress::from(*key));
+            let stake = serai
+              .validator_sets()
+              .total_allocated_stake(set.network)
+              .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"))?;
+          }
+
+          let global_session_info = GlobalSession {
+            // This session starts cosigning after this block, as this block must be cosigned by
+            // the existing validators
+            start_block_number: block_number + 1,
+            sets,
+            keys,
+            stakes,
+            total_stake,
+          };
+          GlobalSessions::set(&mut txn, global_session, &global_session_info);
+          if let Some(ending_global_session) = global_session_for_this_block {
+            GlobalSessionsLastBlock::set(&mut txn, ending_global_session, &block_number);
+          }
+          LatestGlobalSessionIntended::set(&mut txn, &global_session);
+          GlobalSessionsChannel::send(&mut txn, &(global_session, global_session_info));
+        }
+
+        // If there isn't anyone available to cosign this block, meaning it'll never be cosigned,
+        // we flag it as not having any events requiring cosigning so we don't attempt to
+        // sign/require a cosign for it
+        if global_session_for_this_block.is_none() {
+          has_events = HasEvents::No;
+        }
+
         match has_events {
           HasEvents::Notable | HasEvents::NonNotable => {
-            let sets = cosigning_sets_for_block(&self.serai, &block).await?;
+            let global_session_for_this_block = global_session_for_this_block
+              .expect("global session for this block was None but still attempting to cosign it");
+            let global_session_info = GlobalSessions::get(&txn, global_session_for_this_block)
+              .expect("last global session intended wasn't saved to the database");
 
-            // If this is notable, it creates a new global session, which we index into the
-            // database now
-            if has_events == HasEvents::Notable {
-              let sets = cosigning_sets(&self.serai.as_of(block.hash())).await?;
-              let global_session = GlobalSession::new(sets).id();
-              GlobalSessions::set(&mut txn, global_session, &(block_number, block.hash()));
-              if let Some(ending_global_session) = LatestGlobalSessionIntended::get(&txn) {
-                GlobalSessionLastBlock::set(&mut txn, ending_global_session, &block_number);
-              }
-              LatestGlobalSessionIntended::set(&mut txn, &global_session);
-            }
-
-            // If this block doesn't have any cosigners, meaning it'll never be cosigned, we flag it
-            // as not having any events requiring cosigning so we don't attempt to sign/require a
-            // cosign for it
-            if sets.is_empty() {
-              has_events = HasEvents::No;
-            } else {
-              let global_session = GlobalSession::new(sets.clone()).id();
-              // Tell each set of their expectation to cosign this block
-              for set in sets {
-                log::debug!("{:?} will be cosigning block #{block_number}", set);
-                IntendedCosigns::send(
-                  &mut txn,
-                  set,
-                  &CosignIntent {
-                    global_session,
-                    block_number,
-                    block_hash: block.hash(),
-                    notable: has_events == HasEvents::Notable,
-                  },
-                );
-              }
+            // Tell each set of their expectation to cosign this block
+            for set in global_session_info.sets {
+              log::debug!("{:?} will be cosigning block #{block_number}", set);
+              IntendedCosigns::send(
+                &mut txn,
+                set,
+                &CosignIntent {
+                  global_session: global_session_for_this_block,
+                  block_number,
+                  block_hash: block.hash(),
+                  notable: has_events == HasEvents::Notable,
+                },
+              );
             }
           }
           HasEvents::No => {}
         }
+
         // Populate a singular feed with every block's status for the evluator to work off of
-        BlockEvents::send(
-          &mut txn,
-          &(BlockEventData {
-            block_number,
-            parent_hash: block.header.parent_hash.into(),
-            block_hash: block.hash(),
-            has_events,
-          }),
-        );
+        BlockEvents::send(&mut txn, &(BlockEventData { block_number, has_events }));
         // Mark this block as handled, meaning we should scan from the next block moving on
         ScanCosignFrom::set(&mut txn, &(block_number + 1));
         txn.commit();

coordinator/cosign/src/lib.rs

@@ -3,15 +3,16 @@
 #![deny(missing_docs)]
 
 use core::{fmt::Debug, future::Future};
+use std::collections::HashMap;
 
 use blake2::{Digest, Blake2s256};
 
 use borsh::{BorshSerialize, BorshDeserialize};
 
 use serai_client::{
-  primitives::{Amount, NetworkId, SeraiAddress},
+  primitives::{NetworkId, SeraiAddress},
   validator_sets::primitives::{Session, ValidatorSet, KeyPair},
-  Block, Serai, TemporalSerai,
+  Public, Block, Serai, TemporalSerai,
 };
 
 use serai_db::*;
@@ -21,7 +22,10 @@ use serai_task::*;
 mod intend;
 /// The evaluator of the cosigns.
 mod evaluator;
-use evaluator::LatestCosignedBlockNumber;
+/// The task to delay acknowledgement of the cosigns.
+mod delay;
+pub use delay::BROADCAST_FREQUENCY;
+use delay::LatestCosignedBlockNumber;
 
 /// The schnorrkel context to used when signing a cosign.
 pub const COSIGN_CONTEXT: &[u8] = b"serai-cosign";
@@ -45,29 +49,37 @@ pub const COSIGN_CONTEXT: &[u8] = b"serai-cosign";
   cosigning protocol.
 */
 #[derive(Debug, BorshSerialize, BorshDeserialize)]
-struct GlobalSession {
-  cosigners: Vec<ValidatorSet>,
+pub(crate) struct GlobalSession {
+  pub(crate) start_block_number: u64,
+  pub(crate) sets: Vec<ValidatorSet>,
+  pub(crate) keys: HashMap<NetworkId, SeraiAddress>,
+  pub(crate) stakes: HashMap<NetworkId, u64>,
+  pub(crate) total_stake: u64,
 }
 impl GlobalSession {
-  fn new(mut cosigners: Vec<ValidatorSet>) -> Self {
+  fn id(mut cosigners: Vec<ValidatorSet>) -> [u8; 32] {
     cosigners.sort_by_key(|a| borsh::to_vec(a).unwrap());
-    Self { cosigners }
-  }
-  fn id(&self) -> [u8; 32] {
-    Blake2s256::digest(borsh::to_vec(self).unwrap()).into()
+    Blake2s256::digest(borsh::to_vec(&cosigners).unwrap()).into()
   }
 }
 
 create_db! {
   Cosign {
-    // A mapping from a global session's ID to its start block (number, hash).
-    GlobalSessions: (global_session: [u8; 32]) -> (u64, [u8; 32]),
+    // The following are populated by the intend task and used throughout the library
+
+    // An index of Substrate blocks
+    SubstrateBlocks: (block_number: u64) -> [u8; 32],
+    // A mapping from a global session's ID to its relevant information.
+    GlobalSessions: (global_session: [u8; 32]) -> GlobalSession,
     // The last block to be cosigned by a global session.
-    GlobalSessionLastBlock: (global_session: [u8; 32]) -> u64,
+    GlobalSessionsLastBlock: (global_session: [u8; 32]) -> u64,
     // The latest global session intended.
     //
     // This is distinct from the latest global session for which we've evaluated the cosigns for.
     LatestGlobalSessionIntended: () -> [u8; 32],
+
+    // The following are managed by the `intake_cosign` function present in this file
+
     // The latest cosigned block for each network.
     //
     // This will only be populated with cosigns predating or during the most recent global session
@@ -112,15 +124,6 @@ struct CosignIntent {
   notable: bool,
 }
 
-/// The identification of a cosigner.
-#[derive(Clone, Copy, PartialEq, Eq, Debug, BorshSerialize, BorshDeserialize)]
-pub enum Cosigner {
-  /// The network which produced this cosign.
-  ValidatorSet(NetworkId),
-  /// The individual validator which produced this cosign.
-  Validator(SeraiAddress),
-}
-
 /// A cosign.
 #[derive(Clone, PartialEq, Eq, Debug, BorshSerialize, BorshDeserialize)]
 pub struct Cosign {
@@ -131,7 +134,7 @@ pub struct Cosign {
   /// The hash of the block to cosign.
   pub block_hash: [u8; 32],
   /// The actual cosigner.
-  pub cosigner: Cosigner,
+  pub cosigner: NetworkId,
 }
 
 /// A signed cosign.
@@ -189,41 +192,20 @@ async fn keys_for_network(
   Ok(None)
 }
 
-/// Fetch the `ValidatorSet`s used for cosigning as of this block.
-async fn cosigning_sets(serai: &TemporalSerai<'_>) -> Result<Vec<ValidatorSet>, String> {
+/// Fetch the `ValidatorSet`s, and their associated keys, used for cosigning as of this block.
+async fn cosigning_sets(serai: &TemporalSerai<'_>) -> Result<Vec<(ValidatorSet, Public)>, String> {
   let mut sets = Vec::with_capacity(serai_client::primitives::NETWORKS.len());
   for network in serai_client::primitives::NETWORKS {
-    let Some((session, _)) = keys_for_network(serai, network).await? else {
+    let Some((session, keys)) = keys_for_network(serai, network).await? else {
       // If this network doesn't have usable keys, move on
       continue;
     };
 
-    sets.push(ValidatorSet { network, session });
+    sets.push((ValidatorSet { network, session }, keys.0));
   }
   Ok(sets)
 }
 
-/// Fetch the `ValidatorSet`s used for cosigning a block by the block's parent hash.
-async fn cosigning_sets_by_parent_hash(
-  serai: &Serai,
-  parent_hash: [u8; 32],
-) -> Result<Vec<ValidatorSet>, String> {
-  /*
-    If we're cosigning block `n`, it's cosigned by the sets as of block `n-1` (as block `n` may
-    update the sets declared but that update shouldn't take effect until block `n` is cosigned).
-    That's why fetching the cosigning sets for a block by its parent hash is valid.
-  */
-  cosigning_sets(&serai.as_of(parent_hash)).await
-}
-
-/// Fetch the `ValidatorSet`s used for cosigning this block.
-async fn cosigning_sets_for_block(
-  serai: &Serai,
-  block: &Block,
-) -> Result<Vec<ValidatorSet>, String> {
-  cosigning_sets_by_parent_hash(serai, block.header.parent_hash.into()).await
-}
-
 /// 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.
@@ -242,7 +224,6 @@ pub struct Faulted;
 /// The interface to manage cosigning with.
 pub struct Cosigning<D: Db> {
   db: D,
-  serai: Serai,
 }
 impl<D: Db> Cosigning<D> {
   /// Spawn the tasks to intend and evaluate cosigns.
@@ -257,15 +238,20 @@ impl<D: Db> Cosigning<D> {
   ) -> Self {
     let (intend_task, _intend_task_handle) = Task::new();
     let (evaluator_task, evaluator_task_handle) = Task::new();
+    let (delay_task, delay_task_handle) = Task::new();
     tokio::spawn(
-      (intend::CosignIntendTask { db: db.clone(), serai: serai.clone() })
+      (intend::CosignIntendTask { db: db.clone(), serai })
         .continually_run(intend_task, vec![evaluator_task_handle]),
     );
     tokio::spawn(
-      (evaluator::CosignEvaluatorTask { db: db.clone(), serai: serai.clone(), request })
-        .continually_run(evaluator_task, tasks_to_run_upon_cosigning),
+      (evaluator::CosignEvaluatorTask { db: db.clone(), request })
+        .continually_run(evaluator_task, vec![delay_task_handle]),
     );
-    Self { db, serai }
+    tokio::spawn(
+      (delay::CosignDelayTask { db: db.clone() })
+        .continually_run(delay_task, tasks_to_run_upon_cosigning),
+    );
+    Self { db }
   }
 
   /// The latest cosigned block number.
@@ -291,7 +277,7 @@ impl<D: Db> Cosigning<D> {
     cosigns
   }
 
-  /// The cosigns to rebroadcast ever so often.
+  /// The cosigns to rebroadcast every `BROADCAST_FREQUENCY` seconds.
   ///
   /// This will be the most recent cosigns, in case the initial broadcast failed, or the faulty
   /// cosigns, in case of a fault, to induce identification of the fault by others.
@@ -338,59 +324,56 @@ impl<D: Db> Cosigning<D> {
   //
   // Takes `&mut self` as this should only be called once at any given moment.
   // TODO: Don't overload bool here
-  pub async fn intake_cosign(&mut self, signed_cosign: SignedCosign) -> Result<bool, String> {
+  pub fn intake_cosign(&mut self, signed_cosign: &SignedCosign) -> Result<bool, String> {
     let cosign = &signed_cosign.cosign;
+    let network = cosign.cosigner;
 
-    let Cosigner::ValidatorSet(network) = cosign.cosigner else {
-      // TODO
-      // Individually signed cosign despite that protocol not being implemented
-      return Ok(false);
+    // Check our indexed blockchain includes a block with this block number
+    let Some(our_block_hash) = SubstrateBlocks::get(&self.db, cosign.block_number) else {
+      return Ok(true);
     };
+    let faulty = cosign.block_hash != our_block_hash;
 
-    // Check this isn't a dated cosign
-    if let Some(existing) =
-      NetworksLatestCosignedBlock::get(&self.db, cosign.global_session, network)
-    {
-      if existing.cosign.block_number >= cosign.block_number {
-        return Ok(true);
+    // Check this isn't a dated cosign within its global session (as it would be if rebroadcasted)
+    if !faulty {
+      if let Some(existing) =
+        NetworksLatestCosignedBlock::get(&self.db, cosign.global_session, network)
+      {
+        if existing.cosign.block_number >= cosign.block_number {
+          return Ok(true);
+        }
       }
     }
 
-    // Check our finalized (and indexed by intend) blockchain exceeds this block number
-    if cosign.block_number >= intend::ScanCosignFrom::get(&self.db).unwrap_or(0) {
-      return Ok(true);
-    }
-
-    let Some((global_session_start_block_number, global_session_start_block_hash)) =
-      GlobalSessions::get(&self.db, cosign.global_session)
-    else {
+    let Some(global_session) = GlobalSessions::get(&self.db, cosign.global_session) else {
       // Unrecognized global session
       return Ok(true);
     };
-    if cosign.block_number <= global_session_start_block_number {
+
+    // Check the cosigned block number is in range to the global session
+    if cosign.block_number < global_session.start_block_number {
       // Cosign is for a block predating the global session
       return Ok(false);
     }
-    if Some(cosign.block_number) > GlobalSessionLastBlock::get(&self.db, cosign.global_session) {
-      // Cosign is for a block after the last block this global session should have signed
-      return Ok(false);
+    if !faulty {
+      // This prevents a malicious validator set, on the same chain, from producing a cosign after
+      // their final block, replacing their notable cosign
+      if let Some(last_block) = GlobalSessionsLastBlock::get(&self.db, cosign.global_session) {
+        if cosign.block_number > last_block {
+          // Cosign is for a block after the last block this global session should have signed
+          return Ok(false);
+        }
+      }
     }
 
     // Check the cosign's signature
     {
-      let key = match cosign.cosigner {
-        Cosigner::ValidatorSet(network) => {
-          // TODO: Cache this
-          let Some((_session, keys)) =
-            keys_for_network(&self.serai.as_of(global_session_start_block_hash), network).await?
-          else {
-            return Ok(false);
-          };
-
-          keys.0
-        }
-        Cosigner::Validator(signer) => signer.into(),
-      };
+      let key = Public::from({
+        let Some(key) = global_session.keys.get(&network) else {
+          return Ok(false);
+        };
+        *key
+      });
 
       if !signed_cosign.verify_signature(key) {
         return Ok(false);
@@ -402,53 +385,35 @@ impl<D: Db> Cosigning<D> {
 
     let mut txn = self.db.txn();
 
-    let our_block_hash = self
-      .serai
-      .block_hash(cosign.block_number)
-      .await
-      .map_err(|e| format!("{e:?}"))?
-      .ok_or_else(|| "requested hash of a finalized block yet received None".to_string())?;
-    if our_block_hash == cosign.block_hash {
+    if !faulty {
       // If this is for a future global session, we don't acknowledge this cosign at this time
-      if global_session_start_block_number > LatestCosignedBlockNumber::get(&txn).unwrap_or(0) {
+      let latest_cosigned_block_number = LatestCosignedBlockNumber::get(&txn).unwrap_or(0);
+      // This global session starts the block *after* its declaration, so we want to check if the
+      // block declaring it was cosigned
+      if (global_session.start_block_number - 1) > latest_cosigned_block_number {
         drop(txn);
         return Ok(true);
       }
 
-      NetworksLatestCosignedBlock::set(&mut txn, cosign.global_session, network, &signed_cosign);
+      // This is safe as it's in-range and newer, as prior checked since it isn't faulty
+      NetworksLatestCosignedBlock::set(&mut txn, cosign.global_session, network, signed_cosign);
     } else {
       let mut faults = Faults::get(&txn, cosign.global_session).unwrap_or(vec![]);
       // Only handle this as a fault if this set wasn't prior faulty
-      if !faults.iter().any(|cosign| cosign.cosign.cosigner == Cosigner::ValidatorSet(network)) {
+      if !faults.iter().any(|cosign| cosign.cosign.cosigner == network) {
         faults.push(signed_cosign.clone());
         Faults::set(&mut txn, cosign.global_session, &faults);
 
         let mut weight_cosigned = 0;
-        let mut total_weight = 0;
-        for set in cosigning_sets(&self.serai.as_of(global_session_start_block_hash)).await? {
-          let stake = self
-            .serai
-            .as_of(global_session_start_block_hash)
-            .validator_sets()
-            .total_allocated_stake(set.network)
-            .await
-            .map_err(|e| format!("{e:?}"))?
-            .unwrap_or(Amount(0))
-            .0;
-          // Increment total_weight with this set's stake
-          total_weight += stake;
-
-          // Check if this set cosigned this block or not
-          if faults
-            .iter()
-            .any(|cosign| cosign.cosign.cosigner == Cosigner::ValidatorSet(set.network))
-          {
-            weight_cosigned += total_weight
-          }
+        for fault in &faults {
+          let Some(stake) = global_session.stakes.get(&fault.cosign.cosigner) else {
+            Err("cosigner with recognized key didn't have a stake entry saved".to_string())?
+          };
+          weight_cosigned += stake;
         }
 
         // Check if the sum weight means a fault has occurred
-        if weight_cosigned >= ((total_weight * 17) / 100) {
+        if weight_cosigned >= ((global_session.total_stake * 17) / 100) {
           FaultedSession::set(&mut txn, &cosign.global_session);
         }
       }

patches/rocksdb/Cargo.toml

@@ -13,10 +13,10 @@ all-features = true
 rustdoc-args = ["--cfg", "docsrs"]
 
 [dependencies]
-rocksdb = { version = "0.22", default-features = false }
+rocksdb = { version = "0.23", default-features = false, features = ["bindgen-runtime"] }
 
 [features]
-jemalloc = []
+jemalloc = [] # Dropped as this causes a compilation failure on windows
 snappy = ["rocksdb/snappy"]
 lz4 = ["rocksdb/lz4"]
 zstd = ["rocksdb/zstd"]