Make transactions undroppable

coordinator/cosign/src/delay.rs literally demonstrates how we'd need to rewrite
our handling of transactions with this change. It can be cleaned up a bit but
already identifies ergonomic issues. It also doesn't model passing an &mut txn
to an async function, which would also require using the droppable wrapper
struct.

To locally see this build, run

RUSTFLAGS="-Zpanic_abort_tests -C panic=abort" cargo +nightly build -p serai-cosign --all-targets

To locally see this fail to build, run

cargo build -p serai-cosign --all-targets

While it doesn't say which line causes it fail to build, the only distinction
is panic=unwind.

For more context, please see #578.

common/db/src/lib.rs

@@ -30,13 +30,53 @@ pub trait Get {
 /// is undefined. The transaction may block, deadlock, panic, overwrite one of the two values
 /// randomly, or any other action, at time of write or at time of commit.
 #[must_use]
-pub trait DbTxn: Send + Get {
+pub trait DbTxn: Sized + Send + Get {
   /// Write a value to this key.
   fn put(&mut self, key: impl AsRef<[u8]>, value: impl AsRef<[u8]>);
   /// Delete the value from this key.
   fn del(&mut self, key: impl AsRef<[u8]>);
   /// Commit this transaction.
   fn commit(self);
+  /// Close this transaction.
+  ///
+  /// This is equivalent to `Drop` on transactions which can be dropped. This is explicit and works
+  /// with transactions which can't be dropped.
+  fn close(self) {
+    drop(self);
+  }
+}
+
+// Credit for the idea goes to https://jack.wrenn.fyi/blog/undroppable
+pub struct Undroppable<T>(Option<T>);
+impl<T> Drop for Undroppable<T> {
+  fn drop(&mut self) {
+    // Use an assertion at compile time to prevent this code from compiling if generated
+    #[allow(clippy::assertions_on_constants)]
+    const {
+      assert!(false, "Undroppable DbTxn was dropped. Ensure all code paths call commit or close");
+    }
+  }
+}
+impl<T: DbTxn> Get for Undroppable<T> {
+  fn get(&self, key: impl AsRef<[u8]>) -> Option<Vec<u8>> {
+    self.0.as_ref().unwrap().get(key)
+  }
+}
+impl<T: DbTxn> DbTxn for Undroppable<T> {
+  fn put(&mut self, key: impl AsRef<[u8]>, value: impl AsRef<[u8]>) {
+    self.0.as_mut().unwrap().put(key, value);
+  }
+  fn del(&mut self, key: impl AsRef<[u8]>) {
+    self.0.as_mut().unwrap().del(key);
+  }
+  fn commit(mut self) {
+    self.0.take().unwrap().commit();
+    let _ = core::mem::ManuallyDrop::new(self);
+  }
+  fn close(mut self) {
+    drop(self.0.take().unwrap());
+    let _ = core::mem::ManuallyDrop::new(self);
+  }
 }
 
 /// A database supporting atomic transaction.
@@ -51,6 +91,10 @@ pub trait Db: 'static + Send + Sync + Clone + Get {
     let dst_len = u8::try_from(item_dst.len()).unwrap();
     [[db_len].as_ref(), db_dst, [dst_len].as_ref(), item_dst, key.as_ref()].concat()
   }
-  /// Open a new transaction.
-  fn txn(&mut self) -> Self::Transaction<'_>;
+  /// Open a new transaction which may be dropped.
+  fn unsafe_txn(&mut self) -> Self::Transaction<'_>;
+  /// Open a new transaction which must be committed or closed.
+  fn txn(&mut self) -> Undroppable<Self::Transaction<'_>> {
+    Undroppable(Some(self.unsafe_txn()))
+  }
 }

common/db/src/mem.rs

@@ -74,7 +74,7 @@ impl Get for MemDb {
 }
 impl Db for MemDb {
   type Transaction<'a> = MemDbTxn<'a>;
-  fn txn(&mut self) -> MemDbTxn<'_> {
+  fn unsafe_txn(&mut self) -> MemDbTxn<'_> {
     MemDbTxn(self, HashMap::new(), HashSet::new())
   }
 }

common/db/src/parity_db.rs

@@ -37,7 +37,7 @@ impl Get for Arc<ParityDb> {
 }
 impl Db for Arc<ParityDb> {
   type Transaction<'a> = Transaction<'a>;
-  fn txn(&mut self) -> Self::Transaction<'_> {
+  fn unsafe_txn(&mut self) -> Self::Transaction<'_> {
     Transaction(self, vec![])
   }
 }

common/db/src/rocks.rs

@@ -39,7 +39,7 @@ impl<T: ThreadMode> Get for Arc<OptimisticTransactionDB<T>> {
 }
 impl<T: Send + ThreadMode + 'static> Db for Arc<OptimisticTransactionDB<T>> {
   type Transaction<'a> = Transaction<'a, T>;
-  fn txn(&mut self) -> Self::Transaction<'_> {
+  fn unsafe_txn(&mut self) -> Self::Transaction<'_> {
     let mut opts = WriteOptions::default();
     opts.set_sync(true);
     Transaction(self.transaction_opt(&opts, &Default::default()), &**self)

coordinator/cosign/src/delay.rs

@@ -24,6 +24,15 @@ pub(crate) struct CosignDelayTask<D: Db> {
   pub(crate) db: D,
 }
 
+struct AwaitUndroppable<T: DbTxn>(Option<core::mem::ManuallyDrop<Undroppable<T>>>);
+impl<T: DbTxn> Drop for AwaitUndroppable<T> {
+  fn drop(&mut self) {
+    if let Some(mut txn) = self.0.take() {
+      (unsafe { core::mem::ManuallyDrop::take(&mut txn) }).close();
+    }
+  }
+}
+
 impl<D: Db> ContinuallyRan for CosignDelayTask<D> {
   type Error = DoesNotError;
 
@@ -35,14 +44,18 @@ impl<D: Db> ContinuallyRan for CosignDelayTask<D> {
 
         // Receive the next block to mark as cosigned
         let Some((block_number, time_evaluated)) = CosignedBlocks::try_recv(&mut txn) else {
+          txn.close();
           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
+        let mut txn = AwaitUndroppable(Some(core::mem::ManuallyDrop::new(txn)));
         tokio::time::sleep(SystemTime::now().duration_since(time_valid).unwrap_or(Duration::ZERO))
           .await;
+        let mut txn = core::mem::ManuallyDrop::into_inner(txn.0.take().unwrap());
 
         // Set the cosigned block
         LatestCosignedBlockNumber::set(&mut txn, &block_number);

coordinator/cosign/src/evaluator.rs

@@ -87,7 +87,7 @@ impl<D: Db, R: RequestNotableCosigns> ContinuallyRan for CosignEvaluatorTask<D,
       let mut known_cosign = None;
       let mut made_progress = false;
       loop {
-        let mut txn = self.db.txn();
+        let mut txn = self.db.unsafe_txn();
         let Some(BlockEventData { block_number, has_events }) = BlockEvents::try_recv(&mut txn)
         else {
           break;

coordinator/cosign/src/intend.rs

@@ -70,7 +70,7 @@ impl<D: Db> ContinuallyRan for CosignIntendTask<D> {
         self.serai.latest_finalized_block().await.map_err(|e| format!("{e:?}"))?.number();
 
       for block_number in start_block_number ..= latest_block_number {
-        let mut txn = self.db.txn();
+        let mut txn = self.db.unsafe_txn();
 
         let (block, mut has_events) =
           block_has_events_justifying_a_cosign(&self.serai, block_number)

coordinator/cosign/src/lib.rs

@@ -424,7 +424,7 @@ impl<D: Db> Cosigning<D> {
     // Since we verified this cosign's signature, and have a chain sufficiently long, handle the
     // cosign
 
-    let mut txn = self.db.txn();
+    let mut txn = self.db.unsafe_txn();
 
     if !faulty {
       // If this is for a future global session, we don't acknowledge this cosign at this time
@@ -480,3 +480,30 @@ impl<D: Db> Cosigning<D> {
     res
   }
 }
+
+mod tests {
+  use super::*;
+
+  struct RNC;
+  impl RequestNotableCosigns for RNC {
+    /// The error type which may be encountered when requesting notable cosigns.
+    type Error = ();
+
+    /// Request the notable cosigns for this global session.
+    fn request_notable_cosigns(
+      &self,
+      global_session: [u8; 32],
+    ) -> impl Send + Future<Output = Result<(), Self::Error>> {
+      async move { Ok(()) }
+    }
+  }
+  #[tokio::test]
+  async fn test() {
+    let db: serai_db::MemDb = serai_db::MemDb::new();
+    let serai = unsafe { core::mem::transmute(0u64) };
+    let request = RNC;
+    let tasks = vec![];
+    let _ = Cosigning::spawn(db, serai, request, tasks);
+    core::future::pending().await
+  }
+}

substrate/validator-sets/primitives/src/slash_points.rs

@@ -84,12 +84,17 @@ impl Slash {
           - Hours 12-24 are penalized as if they're hours 12-36.
           - Hours 24-36 are penalized as if they're hours 36-96.
           - Hours 36-48 are penalized as if they're hours 96-168.
+
+          /* Commented, see below explanation of why.
           - Hours 48-168 are penalized for 0-2% of stake.
           - 168-336 hours of slashes, for a session only lasting 168 hours, is penalized for 2-10%
             of stake.
 
           This means a validator offline has to be offline for more than two days to start having
           their stake slashed.
+          */
+
+          This means a validator offline for two days will not earn any rewards for that session.
         */
 
         const MULTIPLIERS: [u64; 4] = [1, 2, 5, 6];
@@ -116,7 +121,10 @@ impl Slash {
               }
             })
         };
+        // Ensure the slash never exceeds the amount slashable (due to rounding errors)
+        let reward_slash = reward_slash.min(session_rewards);
 
+        /*
         let slash_points_for_entire_session =
           SESSION_LENGTH.as_secs() / downtime_per_slash_point.as_secs();
 
@@ -159,11 +167,36 @@ impl Slash {
             .checked_div(u128::from(slash_points_for_entire_session))
             .unwrap_or(0)
         };
+        */
 
-        // The penalty is all slashes, but never more than the validator's balance
-        // (handles any rounding errors which may or may not exist)
-        let penalty_u128 =
-          (reward_slash + offline_slash + disruptive_slash).min(allocated_stake + session_rewards);
+        /*
+          We do not slash for being offline/disruptive at this time. Doing so allows an adversary
+          to DoS nodes to not just take them offline, yet also take away their stake. This isn't
+          preferable to the increased incentive to properly maintain a node when the rewards should
+          already be sufficient for that purpose.
+
+          Validators also shouldn't be able to be so disruptive due to their limiting upon
+          disruption *while its ongoing*. Slashes as a post-response, while an arguably worthwhile
+          economic penalty, can never be a response in the moment (as necessary to actually handle
+          the disruption).
+
+          If stake slashing was to be re-enabled, the percentage of stake which is eligible for
+          slashing should be variable to how close we are to losing liveness. This would mean if
+          less than 10% of validators are offline, no stake is slashes. If 10% are, 2% is eligible.
+          If 20% are, 5% is eligible. If 30% are, 10% is eligible.
+
+          (or similar)
+
+          This would mean that a DoS is insufficient to cause a validator to lose their stake.
+          Instead, a coordinated DoS against multiple Serai validators would be needed,
+          strengthening our assumptions.
+        */
+        let offline_slash = 0;
+        let disruptive_slash = 0;
+
+        let stake_slash = (offline_slash + disruptive_slash).min(allocated_stake);
+
+        let penalty_u128 = reward_slash + stake_slash;
         // saturating_into
         Amount(u64::try_from(penalty_u128).unwrap_or(u64::MAX))
       }
@@ -225,6 +258,7 @@ fn test_penalty() {
       Amount(168)
     );
 
+    /*
     // A full week of slash points should slash 2%
     let week_of_slash_points = 14 * twelve_hours_of_slash_points;
     assert_eq!(
@@ -243,6 +277,13 @@ fn test_penalty() {
       Slash::Points(u32::MAX).penalty(validators, Amount(1000), Amount(168)),
       Amount(100 + 168)
     );
+    */
+
+    // Anything greater should still only slash the rewards
+    assert_eq!(
+      Slash::Points(u32::MAX).penalty(validators, Amount(u64::MAX), Amount(168)),
+      Amount(168)
+    );
   }
 }