Ecosystem/serai
Ecosystem/serai
1
0
Fork 0
mirror of https://github.com/serai-dex/serai.git synced 2025-01-22 02:34:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Multisig db (#444)

Browse source 
* implement db macro for processor/multisigs/db.rs

* ran fmt

* cargo +nightly fmt

---------

Co-authored-by: Luke Parker <lukeparker5132@gmail.com>
This commit is contained in:
econsta 2023-11-21 06:40:54 +04:00 committed by GitHub
parent 8634d90b6b
commit 05d8c32be8
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 93 additions and 134 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

188
processor/src/multisigs/db.rs
View file

@ -1,56 +1,39 @@
use core::marker::PhantomData;
use ciphersuite::Ciphersuite; use ciphersuite::Ciphersuite;
pub use serai_db::*; pub use serai_db::*;
use scale::{Encode, Decode}; use scale::{Encode, Decode};
use serai_client::{primitives::Balance, in_instructions::primitives::InInstructionWithBalance}; use serai_client::{primitives::Balance, in_instructions::primitives::InInstructionWithBalance};
use crate::{ use crate::{
Get, Db, Plan, Get, Plan,
networks::{Transaction, Network}, networks::{Transaction, Network},
}; };
#[derive(Debug)] create_db!(
pub struct MultisigsDb<N: Network, D: Db>(PhantomData<N>, PhantomData<D>); MultisigsDb {
impl<N: Network, D: Db> MultisigsDb<N, D> { NextBatchDb: () -> u32,
fn multisigs_key(dst: &'static [u8], key: impl AsRef<[u8]>) -> Vec<u8> { PlanDb: (id: &[u8]) -> Vec<u8>,
D::key(b"MULTISIGS", dst, key) PlansFromScanningDb: (block_number: u64) -> Vec<u8>,
OperatingCostsDb: () -> u64,
ResolvedDb: (tx: &[u8]) -> [u8; 32],
SigningDb: (key: &[u8]) -> Vec<u8>,
ForwardedOutputDb: (balance: Balance) -> Vec<u8>,
DelayedOutputDb: () -> Vec<u8>
} }
);
fn next_batch_key() -> Vec<u8> { impl PlanDb {
Self::multisigs_key(b"next_batch", []) pub fn save_active_plan<N: Network>(
} txn: &mut impl DbTxn,
// Set the next batch ID to use
pub fn set_next_batch_id(txn: &mut D::Transaction<'_>, batch: u32) {
txn.put(Self::next_batch_key(), batch.to_le_bytes());
}
// Get the next batch ID
pub fn next_batch_id<G: Get>(getter: &G) -> u32 {
getter.get(Self::next_batch_key()).map_or(0, |v| u32::from_le_bytes(v.try_into().unwrap()))
}
fn plan_key(id: &[u8]) -> Vec<u8> {
Self::multisigs_key(b"plan", id)
}
fn resolved_key(tx: &[u8]) -> Vec<u8> {
Self::multisigs_key(b"resolved", tx)
}
fn signing_key(key: &[u8]) -> Vec<u8> {
Self::multisigs_key(b"signing", key)
}
pub fn save_active_plan(
txn: &mut D::Transaction<'_>,
key: &[u8], key: &[u8],
block_number: u64, block_number: usize,
plan: &Plan<N>, plan: &Plan<N>,
operating_costs_at_time: u64, operating_costs_at_time: u64,
) { ) {
let id = plan.id(); let id = plan.id();
{ {
let mut signing = txn.get(Self::signing_key(key)).unwrap_or(vec![]); let mut signing = SigningDb::get(txn, key).unwrap_or_default();
// If we've already noted we're signing this, return // If we've already noted we're signing this, return
assert_eq!(signing.len() % 32, 0); assert_eq!(signing.len() % 32, 0);
@ -61,25 +44,25 @@ impl<N: Network, D: Db> MultisigsDb<N, D> {
} }
signing.extend(&id); signing.extend(&id);
txn.put(Self::signing_key(key), id); SigningDb::set(txn, key, &id);
} }
{ {
let mut buf = block_number.to_le_bytes().to_vec(); let mut buf = block_number.to_le_bytes().to_vec();
plan.write(&mut buf).unwrap(); plan.write(&mut buf).unwrap();
buf.extend(&operating_costs_at_time.to_le_bytes()); buf.extend(&operating_costs_at_time.to_le_bytes());
txn.put(Self::plan_key(&id), &buf); Self::set(txn, &id, &buf);
} }
} }
pub fn active_plans<G: Get>(getter: &G, key: &[u8]) -> Vec<(u64, Plan<N>, u64)> { pub fn active_plans<N: Network>(getter: &impl Get, key: &[u8]) -> Vec<(u64, Plan<N>, u64)> {
let signing = getter.get(Self::signing_key(key)).unwrap_or(vec![]); let signing = SigningDb::get(getter, key).unwrap_or_default();
let mut res = vec![]; let mut res = vec![];
assert_eq!(signing.len() % 32, 0); assert_eq!(signing.len() % 32, 0);
for i in 0 .. (signing.len() / 32) { for i in 0 .. (signing.len() / 32) {
let id = &signing[(i * 32) .. ((i + 1) * 32)]; let id = &signing[(i * 32) .. ((i + 1) * 32)];
let buf = getter.get(Self::plan_key(id)).unwrap(); let buf = Self::get(getter, id).unwrap();
let block_number = u64::from_le_bytes(buf[.. 8].try_into().unwrap()); let block_number = u64::from_le_bytes(buf[.. 8].try_into().unwrap());
let plan = Plan::<N>::read::<&[u8]>(&mut &buf[8 ..]).unwrap(); let plan = Plan::<N>::read::<&[u8]>(&mut &buf[8 ..]).unwrap();
@ -87,50 +70,41 @@ impl<N: Network, D: Db> MultisigsDb<N, D> {
let operating_costs = u64::from_le_bytes(buf[(buf.len() - 8) ..].try_into().unwrap()); let operating_costs = u64::from_le_bytes(buf[(buf.len() - 8) ..].try_into().unwrap());
res.push((block_number, plan, operating_costs)); res.push((block_number, plan, operating_costs));
} }
res res
} }
fn operating_costs_key() -> Vec<u8> { pub fn plan_by_key_with_self_change<N: Network>(
Self::multisigs_key(b"operating_costs", []) getter: &impl Get,
}
pub fn take_operating_costs(txn: &mut D::Transaction<'_>) -> u64 {
let existing = txn
.get(Self::operating_costs_key())
.map(|bytes| u64::from_le_bytes(bytes.try_into().unwrap()))
.unwrap_or(0);
txn.del(Self::operating_costs_key());
existing
}
pub fn set_operating_costs(txn: &mut D::Transaction<'_>, amount: u64) {
if amount != 0 {
txn.put(Self::operating_costs_key(), amount.to_le_bytes());
}
}
pub fn resolved_plan<G: Get>(
getter: &G,
tx: <N::Transaction as Transaction<N>>::Id,
) -> Option<[u8; 32]> {
getter.get(Self::resolved_key(tx.as_ref())).map(|id| id.try_into().unwrap())
}
pub fn plan_by_key_with_self_change<G: Get>(
getter: &G,
key: <N::Curve as Ciphersuite>::G, key: <N::Curve as Ciphersuite>::G,
id: [u8; 32], id: [u8; 32],
) -> bool { ) -> bool {
let plan = let plan = Plan::<N>::read::<&[u8]>(&mut &Self::get(getter, &id).unwrap()[8 ..]).unwrap();
Plan::<N>::read::<&[u8]>(&mut &getter.get(Self::plan_key(&id)).unwrap()[8 ..]).unwrap();
assert_eq!(plan.id(), id); assert_eq!(plan.id(), id);
(key == plan.key) && (Some(N::change_address(plan.key)) == plan.change) (key == plan.key) && (Some(N::change_address(plan.key)) == plan.change)
} }
pub fn resolve_plan( }
txn: &mut D::Transaction<'_>,
impl OperatingCostsDb {
pub fn take_operating_costs(txn: &mut impl DbTxn) -> u64 {
let existing = Self::get(txn).unwrap_or_default();
txn.del(Self::key());
existing
}
pub fn set_operating_costs(txn: &mut impl DbTxn, amount: u64) {
if amount != 0 {
Self::set(txn, &amount);
}
}
}
impl ResolvedDb {
pub fn resolve_plan<N: Network>(
txn: &mut impl DbTxn,
key: &[u8], key: &[u8],
plan: [u8; 32], plan: [u8; 32],
resolution: <N::Transaction as Transaction<N>>::Id, resolution: <N::Transaction as Transaction<N>>::Id,
) { ) {
let mut signing = txn.get(Self::signing_key(key)).unwrap_or(vec![]); let mut signing = SigningDb::get(txn, key).unwrap_or_default();
assert_eq!(signing.len() % 32, 0); assert_eq!(signing.len() % 32, 0);
let mut found = false; let mut found = false;
@ -147,17 +121,14 @@ impl<N: Network, D: Db> MultisigsDb<N, D> {
if !found { if !found {
log::warn!("told to finish signing {} yet wasn't actively signing it", hex::encode(plan)); log::warn!("told to finish signing {} yet wasn't actively signing it", hex::encode(plan));
} }
SigningDb::set(txn, key, &signing);
txn.put(Self::signing_key(key), signing); Self::set(txn, resolution.as_ref(), &plan);
txn.put(Self::resolved_key(resolution.as_ref()), plan);
} }
}
fn plans_from_scanning_key(block_number: usize) -> Vec<u8> { impl PlansFromScanningDb {
Self::multisigs_key(b"plans_from_scanning", u32::try_from(block_number).unwrap().to_le_bytes()) pub fn set_plans_from_scanning<N: Network>(
} txn: &mut impl DbTxn,
pub fn set_plans_from_scanning(
txn: &mut D::Transaction<'_>,
block_number: usize, block_number: usize,
plans: Vec<Plan<N>>, plans: Vec<Plan<N>>,
) { ) {
@ -165,14 +136,15 @@ impl<N: Network, D: Db> MultisigsDb<N, D> {
for plan in plans { for plan in plans {
plan.write(&mut buf).unwrap(); plan.write(&mut buf).unwrap();
} }
txn.put(Self::plans_from_scanning_key(block_number), buf); Self::set(txn, block_number.try_into().unwrap(), &buf);
} }
pub fn take_plans_from_scanning(
txn: &mut D::Transaction<'_>, pub fn take_plans_from_scanning<N: Network>(
txn: &mut impl DbTxn,
block_number: usize, block_number: usize,
) -> Option<Vec<Plan<N>>> { ) -> Option<Vec<Plan<N>>> {
let key = Self::plans_from_scanning_key(block_number); let block_number = u64::try_from(block_number).unwrap();
let res = txn.get(&key).map(|plans| { let res = Self::get(txn, block_number).map(|plans| {
let mut plans_ref = plans.as_slice(); let mut plans_ref = plans.as_slice();
let mut res = vec![]; let mut res = vec![];
while !plans_ref.is_empty() { while !plans_ref.is_empty() {
@ -181,56 +153,46 @@ impl<N: Network, D: Db> MultisigsDb<N, D> {
res res
}); });
if res.is_some() { if res.is_some() {
txn.del(key); txn.del(Self::key(block_number));
} }
res res
} }
}
fn forwarded_output_key(balance: Balance) -> Vec<u8> { impl ForwardedOutputDb {
Self::multisigs_key(b"forwarded_output", balance.encode()) pub fn save_forwarded_output(txn: &mut impl DbTxn, instruction: InInstructionWithBalance) {
} let mut existing = Self::get(txn, instruction.balance).unwrap_or_default();
pub fn save_forwarded_output(
txn: &mut D::Transaction<'_>,
instruction: InInstructionWithBalance,
) {
let key = Self::forwarded_output_key(instruction.balance);
let mut existing = txn.get(&key).unwrap_or(vec![]);
existing.extend(instruction.encode()); existing.extend(instruction.encode());
txn.put(key, existing); Self::set(txn, instruction.balance, &existing);
} }
pub fn take_forwarded_output( pub fn take_forwarded_output(
txn: &mut D::Transaction<'_>, txn: &mut impl DbTxn,
balance: Balance, balance: Balance,
) -> Option<InInstructionWithBalance> { ) -> Option<InInstructionWithBalance> {
let key = Self::forwarded_output_key(balance); let outputs = Self::get(txn, balance)?;
let outputs = txn.get(&key)?;
let mut outputs_ref = outputs.as_slice(); let mut outputs_ref = outputs.as_slice();
let res = InInstructionWithBalance::decode(&mut outputs_ref).unwrap(); let res = InInstructionWithBalance::decode(&mut outputs_ref).unwrap();
assert!(outputs_ref.len() < outputs.len()); assert!(outputs_ref.len() < outputs.len());
if outputs_ref.is_empty() { if outputs_ref.is_empty() {
txn.del(&key); txn.del(&Self::key(balance));
} else { } else {
txn.put(&key, outputs_ref); Self::set(txn, balance, &outputs);
} }
Some(res) Some(res)
} }
}
fn delayed_output_keys() -> Vec<u8> { impl DelayedOutputDb {
Self::multisigs_key(b"delayed_outputs", []) pub fn save_delayed_output(txn: &mut impl DbTxn, instruction: InInstructionWithBalance) {
} let mut existing = Self::get(txn).unwrap_or_default();
pub fn save_delayed_output(txn: &mut D::Transaction<'_>, instruction: InInstructionWithBalance) {
let key = Self::delayed_output_keys();
let mut existing = txn.get(&key).unwrap_or(vec![]);
existing.extend(instruction.encode()); existing.extend(instruction.encode());
txn.put(key, existing); Self::set(txn, &existing);
} }
pub fn take_delayed_outputs(txn: &mut D::Transaction<'_>) -> Vec<InInstructionWithBalance> {
let key = Self::delayed_output_keys();
let Some(outputs) = txn.get(&key) else { return vec![] }; pub fn take_delayed_outputs(txn: &mut impl DbTxn) -> Vec<InInstructionWithBalance> {
txn.del(key); let Some(outputs) = Self::get(txn) else { return vec![] };
txn.del(Self::key());
let mut outputs_ref = outputs.as_slice(); let mut outputs_ref = outputs.as_slice();
let mut res = vec![]; let mut res = vec![];

39
processor/src/multisigs/mod.rs
View file

@ -26,7 +26,7 @@ pub mod scanner;
use scanner::{ScannerEvent, ScannerHandle, Scanner}; use scanner::{ScannerEvent, ScannerHandle, Scanner};
mod db; mod db;
use db::MultisigsDb; use db::*;
#[cfg(not(test))] #[cfg(not(test))]
mod scheduler; mod scheduler;
@ -174,9 +174,7 @@ impl<D: Db, N: Network> MultisigManager<D, N> {
// Load any TXs being actively signed // Load any TXs being actively signed
let key = key.to_bytes(); let key = key.to_bytes();
for (block_number, plan, operating_costs) in for (block_number, plan, operating_costs) in PlanDb::active_plans::<N>(raw_db, key.as_ref()) {
MultisigsDb::<N, D>::active_plans(raw_db, key.as_ref())
{
let block_number = block_number.try_into().unwrap(); let block_number = block_number.try_into().unwrap();
let id = plan.id(); let id = plan.id();
@ -557,12 +555,12 @@ impl<D: Db, N: Network> MultisigManager<D, N> {
} }
OutputType::Change => { OutputType::Change => {
// If the TX containing this output resolved an Eventuality... // If the TX containing this output resolved an Eventuality...
if let Some(plan) = MultisigsDb::<N, D>::resolved_plan(txn, output.tx_id()) { if let Some(plan) = ResolvedDb::get(txn, output.tx_id().as_ref()) {
// And the Eventuality had change... // And the Eventuality had change...
// We need this check as Eventualities have a race condition and can't be relied // We need this check as Eventualities have a race condition and can't be relied
// on, as extensively detailed above. Eventualities explicitly with change do have // on, as extensively detailed above. Eventualities explicitly with change do have
// a safe timing window however // a safe timing window however
if MultisigsDb::<N, D>::plan_by_key_with_self_change( if PlanDb::plan_by_key_with_self_change::<N>(
txn, txn,
// Pass the key so the DB checks the Plan's key is this multisig's, preventing a // Pass the key so the DB checks the Plan's key is this multisig's, preventing a
// potential issue where the new multisig creates a Plan with change *and a // potential issue where the new multisig creates a Plan with change *and a
@ -608,7 +606,7 @@ impl<D: Db, N: Network> MultisigManager<D, N> {
block_number block_number
{ {
// Load plans crated when we scanned the block // Load plans crated when we scanned the block
plans = MultisigsDb::<N, D>::take_plans_from_scanning(txn, block_number).unwrap(); plans = PlansFromScanningDb::take_plans_from_scanning::<N>(txn, block_number).unwrap();
for plan in &plans { for plan in &plans {
plans_from_scanning.insert(plan.id()); plans_from_scanning.insert(plan.id());
} }
@ -719,12 +717,12 @@ impl<D: Db, N: Network> MultisigManager<D, N> {
let key_bytes = key.to_bytes(); let key_bytes = key.to_bytes();
let (tx, post_fee_branches) = { let (tx, post_fee_branches) = {
let running_operating_costs = MultisigsDb::<N, D>::take_operating_costs(txn); let running_operating_costs = OperatingCostsDb::take_operating_costs(txn);
MultisigsDb::<N, D>::save_active_plan( PlanDb::save_active_plan::<N>(
txn, txn,
key_bytes.as_ref(), key_bytes.as_ref(),
block_number.try_into().unwrap(), block_number,
&plan, &plan,
running_operating_costs, running_operating_costs,
); );
@ -752,7 +750,7 @@ impl<D: Db, N: Network> MultisigManager<D, N> {
operating_costs += running_operating_costs; operating_costs += running_operating_costs;
} }
MultisigsDb::<N, D>::set_operating_costs(txn, operating_costs); OperatingCostsDb::set_operating_costs(txn, operating_costs);
(tx, post_fee_branches) (tx, post_fee_branches)
}; };
@ -824,8 +822,7 @@ impl<D: Db, N: Network> MultisigManager<D, N> {
continue; continue;
} }
if let Some(instruction) = if let Some(instruction) = ForwardedOutputDb::take_forwarded_output(txn, output.balance())
MultisigsDb::<N, D>::take_forwarded_output(txn, output.balance())
{ {
instructions.push(instruction); instructions.push(instruction);
} }
@ -876,7 +873,7 @@ impl<D: Db, N: Network> MultisigManager<D, N> {
// letting it die out // letting it die out
if let Some(tx) = &tx { if let Some(tx) = &tx {
instruction.balance.amount.0 -= tx.0.fee(); instruction.balance.amount.0 -= tx.0.fee();
MultisigsDb::<N, D>::save_forwarded_output(txn, instruction); ForwardedOutputDb::save_forwarded_output(txn, instruction);
} }
} else if let Some(refund_to) = refund_to { } else if let Some(refund_to) = refund_to {
if let Ok(refund_to) = refund_to.consume().try_into() { if let Ok(refund_to) = refund_to.consume().try_into() {
@ -925,7 +922,7 @@ impl<D: Db, N: Network> MultisigManager<D, N> {
if Some(output.key()) == self.new.as_ref().map(|new| new.key) { if Some(output.key()) == self.new.as_ref().map(|new| new.key) {
match step { match step {
RotationStep::UseExisting => { RotationStep::UseExisting => {
MultisigsDb::<N, D>::save_delayed_output(txn, instruction); DelayedOutputDb::save_delayed_output(txn, instruction);
continue; continue;
} }
RotationStep::NewAsChange | RotationStep::NewAsChange |
@ -940,7 +937,7 @@ impl<D: Db, N: Network> MultisigManager<D, N> {
// Save the plans created while scanning // Save the plans created while scanning
// TODO: Should we combine all of these plans to reduce the fees incurred from their // TODO: Should we combine all of these plans to reduce the fees incurred from their
// execution? They're refunds and forwards. Neither should need isolate Plan/Eventualities. // execution? They're refunds and forwards. Neither should need isolate Plan/Eventualities.
MultisigsDb::<N, D>::set_plans_from_scanning(txn, block_number, plans); PlansFromScanningDb::set_plans_from_scanning(txn, block_number, plans);
// If any outputs were delayed, append them into this block // If any outputs were delayed, append them into this block
match step { match step {
@ -948,13 +945,13 @@ impl<D: Db, N: Network> MultisigManager<D, N> {
RotationStep::NewAsChange | RotationStep::NewAsChange |
RotationStep::ForwardFromExisting | RotationStep::ForwardFromExisting |
RotationStep::ClosingExisting => { RotationStep::ClosingExisting => {
instructions.extend(MultisigsDb::<N, D>::take_delayed_outputs(txn)); instructions.extend(DelayedOutputDb::take_delayed_outputs(txn));
} }
} }
let mut block_hash = [0; 32]; let mut block_hash = [0; 32];
block_hash.copy_from_slice(block.as_ref()); block_hash.copy_from_slice(block.as_ref());
let mut batch_id = MultisigsDb::<N, D>::next_batch_id(txn); let mut batch_id = NextBatchDb::get(txn).unwrap_or_default();
// start with empty batch // start with empty batch
let mut batches = vec![Batch { let mut batches = vec![Batch {
@ -987,7 +984,7 @@ impl<D: Db, N: Network> MultisigManager<D, N> {
} }
// Save the next batch ID // Save the next batch ID
MultisigsDb::<N, D>::set_next_batch_id(txn, batch_id + 1); NextBatchDb::set(txn, &(batch_id + 1));
( (
block_number, block_number,
@ -1006,7 +1003,7 @@ impl<D: Db, N: Network> MultisigManager<D, N> {
// within the block. Unknown Eventualities may have their Completed events emitted after // within the block. Unknown Eventualities may have their Completed events emitted after
// ScannerEvent::Block however. // ScannerEvent::Block however.
ScannerEvent::Completed(key, block_number, id, tx) => { ScannerEvent::Completed(key, block_number, id, tx) => {
MultisigsDb::<N, D>::resolve_plan(txn, &key, id, tx.id()); ResolvedDb::resolve_plan::<N>(txn, &key, id, tx.id());
(block_number, MultisigEvent::Completed(key, id, tx)) (block_number, MultisigEvent::Completed(key, id, tx))
} }
}; };
@ -1028,7 +1025,7 @@ impl<D: Db, N: Network> MultisigManager<D, N> {
let scheduler_is_empty = closing && existing.scheduler.empty(); let scheduler_is_empty = closing && existing.scheduler.empty();
// Nothing is still being signed // Nothing is still being signed
let no_active_plans = scheduler_is_empty && let no_active_plans = scheduler_is_empty &&
MultisigsDb::<N, D>::active_plans(txn, existing.key.to_bytes().as_ref()).is_empty(); PlanDb::active_plans::<N>(txn, existing.key.to_bytes().as_ref()).is_empty();
self self
.scanner .scanner