diff --git a/server/Cargo.lock b/server/Cargo.lock index b8ee5e4..d08434b 100644 --- a/server/Cargo.lock +++ b/server/Cargo.lock @@ -674,6 +674,12 @@ dependencies = [ "libc", ] +[[package]] +name = "anstyle" +version = "1.0.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" + [[package]] name = "anyhow" version = "1.0.102" @@ -767,6 +773,7 @@ dependencies = [ "kameo", "kameo_actors", "ml-dsa", + "mockall", "mutants", "pem", "proptest", @@ -1983,6 +1990,12 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "downcast" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1435fa1053d8b2fbbe9be7e97eca7f33d37b28409959813daefc1446a14247f1" + [[package]] name = "downcast-rs" version = "2.0.2" @@ -2235,6 +2248,15 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fragile" +version = "2.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8878864ba14bb86e818a412bfd6f18f9eabd4ec0f008a28e8f7eb61db532fcf9" +dependencies = [ + "futures-core", +] + [[package]] name = "fs_extra" version = "1.3.0" @@ -3352,6 +3374,32 @@ dependencies = [ "zeroize", ] +[[package]] +name = "mockall" +version = "0.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1a6ceddfe3ce334925e96bf420fdb2dcee5bed6c632a168ece622676dadeaf8a" +dependencies = [ + "cfg-if", + "downcast", + "fragile", + "mockall_derive", + "predicates", + "predicates-tree", +] + +[[package]] +name = "mockall_derive" +version = "0.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9cfe16fbe8a314aeec0b861ac24e60b1e123e97634bab045475b9d6a18416fd8" +dependencies = [ + "cfg-if", + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "module-lattice" version = "0.2.2" @@ -3785,6 +3833,32 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "predicates" +version = "3.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ada8f2932f28a27ee7b70dd6c1c39ea0675c55a36879ab92f3a715eaa1e63cfe" +dependencies = [ + "anstyle", + "predicates-core", +] + +[[package]] +name = "predicates-core" +version = "1.0.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cad38746f3166b4031b1a0d39ad9f954dd291e7854fcc0eed52ee41a0b50d144" + +[[package]] +name = "predicates-tree" +version = "1.0.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0de1b847b39c8131db0467e9df1ff60e6d0562ab8e9a16e568ad0fdb372e2f2" +dependencies = [ + "predicates-core", + "termtree", +] + [[package]] name = "prettyplease" version = "0.2.37" @@ -5171,6 +5245,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "termtree" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f50febec83f5ee1df3015341d8bd429f2d1cc62bcba7ea2076759d315084683" + [[package]] name = "test-log" version = "0.2.20" diff --git a/server/crates/arbiter-server/Cargo.toml b/server/crates/arbiter-server/Cargo.toml index ee58227..b17e739 100644 --- a/server/crates/arbiter-server/Cargo.toml +++ b/server/crates/arbiter-server/Cargo.toml @@ -59,6 +59,7 @@ proptest = "1.11.0" rstest.workspace = true test-log = { version = "0.2", default-features = false, features = ["trace"] } ml-dsa.workspace = true +mockall = "0.15.0" [lib] doctest = false diff --git a/server/crates/arbiter-server/src/actors/proposal_manager.rs b/server/crates/arbiter-server/src/actors/proposal_manager.rs index 62b741a..03a6938 100644 --- a/server/crates/arbiter-server/src/actors/proposal_manager.rs +++ b/server/crates/arbiter-server/src/actors/proposal_manager.rs @@ -1,35 +1,34 @@ use crate::{ - actors::proposal_manager::events::ProposalApproved, + actors::proposal_manager::{ + events::ProposalApproved, + store::{DieselProposalStore, ProposalStore, Tally}, + }, crypto::governance, db::{ self, - functions::unixepoch, models::{ - NewProposal, NewProposalVote, NewRecoveryProposalVote, NewRecoveryWakeupRequest, - OperatorIdentityId, Proposal, ProposalId, ProposalStatus, RecoveryOperatorIdentityId, - SqliteTimestamp, + NewProposalVote, NewRecoveryProposalVote, OperatorIdentityId, Proposal, ProposalId, + ProposalStatus, RecoveryOperatorIdentityId, SqliteTimestamp, }, proposal::{ProposalKind, ProposalKindTag}, - schema, }, }; use chrono::Utc; -use diesel::{ - ExpressionMethods as _, QueryDsl, - dsl::{exists, select}, -}; -use diesel_async::{AsyncConnection as _, RunQueryDsl}; use kameo::{Actor, actor::ActorRef, messages}; use kameo_actors::message_bus::{MessageBus, Publish}; -use std::collections::HashMap; -use strum::IntoDiscriminant as _; +use std::sync::Arc; use tracing::warn; pub mod events; +pub mod store; pub const DEFAULT_TTL_SECS: u32 = 7 * 24 * 60 * 60; // 7 days pub const MAX_TTL_SECS: u32 = DEFAULT_TTL_SECS; +/// Recovery operators stay asleep for this long after a wake-up is requested, so the other +/// operators have time to dispute it (§3.6). +const WAKEUP_DELAY_SECS: i32 = 14 * 24 * 60 * 60; // 14 days + #[derive(Debug, Clone, PartialEq, Eq)] pub enum VoteOutcome { Pending, @@ -81,13 +80,21 @@ pub struct ProposalSummary { #[derive(Actor)] pub struct ProposalManager { - pub(crate) db: db::DatabasePool, + pub(crate) store: Arc, pub(crate) events: ActorRef, } impl ProposalManager { - pub const fn new(db: db::DatabasePool, events: ActorRef) -> Self { - Self { db, events } + pub fn new(db: db::DatabasePool, events: ActorRef) -> Self { + Self::with_store(Arc::new(DieselProposalStore::new(db)), events) + } + + /// Builds the actor over an arbitrary store, so tests can supply a mock. + pub(crate) const fn with_store( + store: Arc, + events: ActorRef, + ) -> Self { + Self { store, events } } } @@ -107,98 +114,18 @@ impl ProposalManager { let expires_at = SqliteTimestamp::from(Utc::now() + chrono::Duration::seconds(i64::from(ttl))); - let id: ProposalId = self - .db - .get() - .await? - .transaction(async |conn| { - let id: ProposalId = diesel::insert_into(schema::proposal::table) - .values(&NewProposal { - kind: kind.discriminant(), - initiator_id, - expires_at, - }) - .returning(schema::proposal::id) - .get_result(conn) - .await?; - db::proposal::insert_kind(conn, id, &kind).await?; - Ok::<_, diesel::result::Error>(id) - }) - .await?; - - Ok(id) + self.store.create(kind, initiator_id, expires_at).await } #[message] pub async fn query_pending(&mut self, operator_id: OperatorIdentityId) -> Vec { - #[expect( - clippy::cast_possible_truncation, - clippy::as_conversions, - reason = "fixme! #84; this will break in 2038" - )] - let now_ts = Utc::now().timestamp() as i32; - - let Ok(mut conn) = self.db.get().await else { - warn!("query_pending: failed to acquire DB connection"); - return vec![]; - }; - - let voted_ids: Vec = schema::proposal_vote::table - .filter(schema::proposal_vote::operator_id.eq(operator_id)) - .select(schema::proposal_vote::proposal_id) - .load(&mut conn) + self.store + .pending_for(operator_id) .await - .unwrap_or_default(); - - let proposals: Vec = schema::proposal::table - .filter(schema::proposal::status.eq(ProposalStatus::Pending)) - .filter(schema::proposal::expires_at.gt(now_ts)) - .filter(diesel::dsl::not(schema::proposal::id.eq_any(&voted_ids))) - .load(&mut conn) - .await - .unwrap_or_default(); - - let ids: Vec = proposals.iter().map(|p| p.id).collect(); - let tallies: Vec<(ProposalId, bool, i64)> = schema::proposal_vote::table - .filter(schema::proposal_vote::proposal_id.eq_any(&ids)) - .group_by(( - schema::proposal_vote::proposal_id, - schema::proposal_vote::approve, - )) - .select(( - schema::proposal_vote::proposal_id, - schema::proposal_vote::approve, - diesel::dsl::count_star(), - )) - .load(&mut conn) - .await - .unwrap_or_default(); - - let mut by_proposal: HashMap = HashMap::new(); - for (proposal_id, approve, count) in tallies { - let entry = by_proposal.entry(proposal_id).or_insert((0, 0)); - if approve { - entry.0 += count; - } else { - entry.1 += count; - } - } - - proposals - .into_iter() - .map(|p| { - let (approve_count, reject_count) = - by_proposal.get(&p.id).copied().unwrap_or((0, 0)); - ProposalSummary { - id: p.id, - kind: p.kind, - initiator_id: p.initiator_id, - expires_at: p.expires_at, - approve_count, - reject_count, - } + .unwrap_or_else(|e| { + warn!(?e, "query_pending failed"); + vec![] }) - .collect() } #[message] @@ -209,141 +136,35 @@ impl ProposalManager { approve: bool, signature: Vec, ) -> Result { - let mut conn = self.db.get().await?; + let proposal = self.store.load(proposal_id).await?; - // Load proposal — must exist - let proposal: Proposal = schema::proposal::table - .find(proposal_id) - .first(&mut conn) - .await - .map_err(|e| match e { - diesel::result::Error::NotFound => Error::ProposalNotFound, - other => Error::DatabaseQuery(other), - })?; - - // Check for duplicate vote before status check so AlreadyVoted takes priority - let already_voted: bool = select(exists( - schema::proposal_vote::table - .filter(schema::proposal_vote::proposal_id.eq(proposal_id)) - .filter(schema::proposal_vote::operator_id.eq(operator_id)), - )) - .get_result(&mut conn) - .await?; - if already_voted { + // Checked before the status check so AlreadyVoted takes priority. + if self.store.has_voted(proposal_id, operator_id).await? { return Err(Error::AlreadyVoted); } - if proposal.status != ProposalStatus::Pending { - return Err(Error::ProposalNotPending); - } + Self::check_votable(&proposal)?; - if proposal.expires_at.0 <= Utc::now() { - return Err(Error::ProposalExpired); - } - - // Load operator public key from operator_identity - let pubkey_bytes: Vec = schema::operator_identity::table - .find(operator_id) - .select(schema::operator_identity::public_key) - .first(&mut conn) - .await - .map_err(|e| match e { - diesel::result::Error::NotFound => Error::OperatorNotFound, - other => Error::DatabaseQuery(other), - })?; - - governance::verify_vote(&pubkey_bytes, proposal_id, approve, &signature) + let public_key = self.store.operator_public_key(operator_id).await?; + governance::verify_vote(&public_key, proposal_id, approve, &signature) .map_err(|_| Error::InvalidSignature)?; - // Insert vote - diesel::insert_into(schema::proposal_vote::table) - .values(&NewProposalVote { + self.store + .record_vote(NewProposalVote { proposal_id, operator_id, approve, signature, }) - .execute(&mut conn) .await?; - // Quorum check - let total_operators: i64 = schema::operator_identity::table - .count() - .get_result(&mut conn) - .await?; - let recovery_active = Self::is_recovery_active_conn(&mut conn).await?; - let total_recovery: i64 = if recovery_active { - schema::recovery_operator_identity::table - .count() - .get_result(&mut conn) - .await? - } else { - 0 - }; - #[expect( - clippy::cast_possible_truncation, - clippy::cast_sign_loss, - clippy::as_conversions, - reason = "operator count is always a small positive integer" - )] - let threshold = if proposal.kind.requires_full_quorum() { - // §3.3: key-rotation proposals require every eligible voter to approve - // §3.5: when recovery is active, recovery operators also vote on replace_operator - (total_operators + total_recovery) as usize - } else { - crate::crypto::shamir::shamir_threshold(total_operators as usize) - }; - - let ordinary_approve: i64 = schema::proposal_vote::table - .filter(schema::proposal_vote::proposal_id.eq(proposal_id)) - .filter(schema::proposal_vote::approve.eq(true)) - .count() - .get_result(&mut conn) - .await?; - let recovery_approve: i64 = schema::recovery_proposal_vote::table - .filter(schema::recovery_proposal_vote::proposal_id.eq(proposal_id)) - .filter(schema::recovery_proposal_vote::approve.eq(true)) - .count() - .get_result(&mut conn) - .await?; - let approve_count = ordinary_approve + recovery_approve; - - let ordinary_reject: i64 = schema::proposal_vote::table - .filter(schema::proposal_vote::proposal_id.eq(proposal_id)) - .filter(schema::proposal_vote::approve.eq(false)) - .count() - .get_result(&mut conn) - .await?; - let recovery_reject: i64 = schema::recovery_proposal_vote::table - .filter(schema::recovery_proposal_vote::proposal_id.eq(proposal_id)) - .filter(schema::recovery_proposal_vote::approve.eq(false)) - .count() - .get_result(&mut conn) - .await?; - let reject_count = ordinary_reject + recovery_reject; - - #[expect( - clippy::cast_possible_wrap, - clippy::as_conversions, - reason = "threshold is derived from operator count, always fits i64" - )] - let threshold_i64 = threshold as i64; - - if approve_count >= threshold_i64 { - self.announce_approval(&mut conn, &proposal).await?; - return Ok(VoteOutcome::Approved); + let mut tally = self.store.tally(proposal_id).await?; + // §3.5: recovery operators only join the electorate once they are awake. + if !self.store.is_recovery_active().await? { + tally.total_recovery = 0; } - let total_eligible = total_operators + total_recovery; - if reject_count > total_eligible - threshold_i64 { - diesel::update(schema::proposal::table.find(proposal_id)) - .set(schema::proposal::status.eq(ProposalStatus::Rejected)) - .execute(&mut conn) - .await?; - return Ok(VoteOutcome::Rejected); - } - - Ok(VoteOutcome::Pending) + self.settle(&proposal, &tally).await } /// §3.6: Any ordinary operator may request recovery wake-up. @@ -353,17 +174,10 @@ impl ProposalManager { &mut self, operator_id: OperatorIdentityId, ) -> Result<(), Error> { - let mut conn = self.db.get().await?; - if Self::has_uncancelled_wakeup(&mut conn).await? { + if self.store.has_uncancelled_wakeup().await? { return Err(Error::WakeupAlreadyPending); } - diesel::insert_into(schema::recovery_wakeup_request::table) - .values(&NewRecoveryWakeupRequest { - requested_by: operator_id, - }) - .execute(&mut conn) - .await?; - Ok(()) + self.store.request_wakeup(operator_id).await } /// §3.6: Any ordinary operator may cancel a pending wake-up request. @@ -373,19 +187,11 @@ impl ProposalManager { &mut self, operator_id: OperatorIdentityId, ) -> Result<(), Error> { - let mut conn = self.db.get().await?; - let rows_updated = diesel::update(schema::recovery_wakeup_request::table) - .filter(schema::recovery_wakeup_request::cancelled_at.is_null()) - .set(( - schema::recovery_wakeup_request::cancelled_by.eq(Some(operator_id)), - schema::recovery_wakeup_request::cancelled_at.eq(Some(SqliteTimestamp::now())), - )) - .execute(&mut conn) - .await?; - if rows_updated == 0 { - return Err(Error::NoActiveWakeup); + if self.store.cancel_wakeup(operator_id).await? { + Ok(()) + } else { + Err(Error::NoActiveWakeup) } - Ok(()) } /// §3.5: Recovery operators may only vote on operator replacement proposals. @@ -398,151 +204,106 @@ impl ProposalManager { approve: bool, signature: Vec, ) -> Result { - let mut conn = self.db.get().await?; - - let proposal: Proposal = schema::proposal::table - .find(proposal_id) - .first(&mut conn) - .await - .map_err(|e| match e { - diesel::result::Error::NotFound => Error::ProposalNotFound, - other => Error::DatabaseQuery(other), - })?; + let proposal = self.store.load(proposal_id).await?; if proposal.kind != ProposalKindTag::ReplaceOperator { return Err(Error::NotAllowedForRecoveryOperator); } - if !Self::is_recovery_active_conn(&mut conn).await? { + if !self.store.is_recovery_active().await? { return Err(Error::RecoveryNotActive); } - let already_voted: bool = select(exists( - schema::recovery_proposal_vote::table - .filter(schema::recovery_proposal_vote::proposal_id.eq(proposal_id)) - .filter( - schema::recovery_proposal_vote::recovery_operator_id.eq(recovery_operator_id), - ), - )) - .get_result(&mut conn) - .await?; - if already_voted { + if self + .store + .has_recovery_voted(proposal_id, recovery_operator_id) + .await? + { return Err(Error::AlreadyVoted); } - if proposal.status != ProposalStatus::Pending { - return Err(Error::ProposalNotPending); - } + Self::check_votable(&proposal)?; - if proposal.expires_at.0 <= Utc::now() { - return Err(Error::ProposalExpired); - } - - let pubkey_bytes: Vec = schema::recovery_operator_identity::table - .find(recovery_operator_id) - .select(schema::recovery_operator_identity::public_key) - .first(&mut conn) - .await - .map_err(|e| match e { - diesel::result::Error::NotFound => Error::OperatorNotFound, - other => Error::DatabaseQuery(other), - })?; - - governance::verify_vote(&pubkey_bytes, proposal_id, approve, &signature) + let public_key = self + .store + .recovery_operator_public_key(recovery_operator_id) + .await?; + governance::verify_vote(&public_key, proposal_id, approve, &signature) .map_err(|_| Error::InvalidSignature)?; - diesel::insert_into(schema::recovery_proposal_vote::table) - .values(&NewRecoveryProposalVote { + self.store + .record_recovery_vote(NewRecoveryProposalVote { proposal_id, recovery_operator_id, approve, signature, }) - .execute(&mut conn) .await?; - // Quorum: all ordinary + all recovery operators must approve (§3.3 + §3.5) - let total_ordinary: i64 = schema::operator_identity::table - .count() - .get_result(&mut conn) - .await?; - let total_recovery: i64 = schema::recovery_operator_identity::table - .count() - .get_result(&mut conn) - .await?; - let threshold_i64 = total_ordinary + total_recovery; - - let ordinary_approve: i64 = schema::proposal_vote::table - .filter(schema::proposal_vote::proposal_id.eq(proposal_id)) - .filter(schema::proposal_vote::approve.eq(true)) - .count() - .get_result(&mut conn) - .await?; - let recovery_approve: i64 = schema::recovery_proposal_vote::table - .filter(schema::recovery_proposal_vote::proposal_id.eq(proposal_id)) - .filter(schema::recovery_proposal_vote::approve.eq(true)) - .count() - .get_result(&mut conn) - .await?; - let approve_count = ordinary_approve + recovery_approve; - - if approve_count >= threshold_i64 { - self.announce_approval(&mut conn, &proposal).await?; - return Ok(VoteOutcome::Approved); - } - - let recovery_reject: i64 = schema::recovery_proposal_vote::table - .filter(schema::recovery_proposal_vote::proposal_id.eq(proposal_id)) - .filter(schema::recovery_proposal_vote::approve.eq(false)) - .count() - .get_result(&mut conn) - .await?; - let ordinary_reject: i64 = schema::proposal_vote::table - .filter(schema::proposal_vote::proposal_id.eq(proposal_id)) - .filter(schema::proposal_vote::approve.eq(false)) - .count() - .get_result(&mut conn) - .await?; - let reject_count = ordinary_reject + recovery_reject; - - if reject_count > threshold_i64 - approve_count - reject_count { - diesel::update(schema::proposal::table.find(proposal_id)) - .set(schema::proposal::status.eq(ProposalStatus::Rejected)) - .execute(&mut conn) - .await?; - return Ok(VoteOutcome::Rejected); - } - - Ok(VoteOutcome::Pending) + let tally = self.store.tally(proposal_id).await?; + self.settle(&proposal, &tally).await } } impl ProposalManager { - const WAKEUP_DELAY_SECS: i32 = 14 * 24 * 60 * 60; - - /// Returns true when an uncancelled wakeup request has passed the 14-day dispute window. - async fn is_recovery_active_conn(conn: &mut db::DatabaseConnection) -> Result { - select(exists( - schema::recovery_wakeup_request::table - .filter(schema::recovery_wakeup_request::cancelled_at.is_null()) - .filter( - schema::recovery_wakeup_request::requested_at - .le(unixepoch("now") - Self::WAKEUP_DELAY_SECS), - ), - )) - .get_result(conn) - .await - .map_err(Error::from) + /// A vote only counts while the proposal is still open. + fn check_votable(proposal: &Proposal) -> Result<(), Error> { + if proposal.status != ProposalStatus::Pending { + return Err(Error::ProposalNotPending); + } + if proposal.expires_at.0 <= Utc::now() { + return Err(Error::ProposalExpired); + } + Ok(()) } - /// Returns true when there is any uncancelled wakeup request (pending or active). - async fn has_uncancelled_wakeup(conn: &mut db::DatabaseConnection) -> Result { - select(exists(schema::recovery_wakeup_request::table.filter( - schema::recovery_wakeup_request::cancelled_at.is_null(), - ))) - .get_result(conn) - .await - .map_err(Error::from) + /// Pure quorum arithmetic — no I/O, so the rules can be tested directly (§3.3). + /// + /// A proposal is rejected once approval has become unreachable: even if every voter + /// who has not spoken yet approved, the threshold could not be met. + #[must_use] + pub(crate) const fn evaluate_quorum(tally: &Tally, requires_full_quorum: bool) -> VoteOutcome { + let total_eligible = tally.total_ordinary + tally.total_recovery; + + #[expect( + clippy::cast_possible_truncation, + clippy::cast_sign_loss, + clippy::cast_possible_wrap, + clippy::as_conversions, + reason = "operator counts are always small positive integers" + )] + // §3.3: key-rotation proposals require every eligible voter to approve. + // §3.5: when recovery is active, recovery operators are eligible too. + let threshold: i64 = if requires_full_quorum { + total_eligible + } else { + crate::crypto::shamir::shamir_threshold(tally.total_ordinary as usize) as i64 + }; + + if tally.approve >= threshold { + VoteOutcome::Approved + } else if tally.reject > total_eligible - threshold { + VoteOutcome::Rejected + } else { + VoteOutcome::Pending + } + } + + /// Applies the quorum rules to a fresh tally and records whatever they decide. + async fn settle(&self, proposal: &Proposal, tally: &Tally) -> Result { + let outcome = Self::evaluate_quorum(tally, proposal.kind.requires_full_quorum()); + + match outcome { + VoteOutcome::Approved => self.announce_approval(proposal).await?, + VoteOutcome::Rejected => { + self.store + .set_status(proposal.id, ProposalStatus::Rejected) + .await?; + } + VoteOutcome::Pending => {} + } + + Ok(outcome) } /// Marks the proposal approved and hands the outcome to whoever owns that kind. @@ -550,17 +311,12 @@ impl ProposalManager { /// The outcome is published, not executed: this actor coordinates voting and nothing /// else. Executors subscribe on the bus, so a vote is answered once the quorum is /// recorded rather than once the effect has landed. - async fn announce_approval( - &self, - conn: &mut db::DatabaseConnection, - proposal: &Proposal, - ) -> Result<(), Error> { - diesel::update(schema::proposal::table.find(proposal.id)) - .set(schema::proposal::status.eq(ProposalStatus::Approved)) - .execute(conn) + async fn announce_approval(&self, proposal: &Proposal) -> Result<(), Error> { + self.store + .set_status(proposal.id, ProposalStatus::Approved) .await?; - let kind = db::proposal::load_kind(conn, proposal.id, proposal.kind).await?; + let kind = self.store.load_kind(proposal.id, proposal.kind).await?; let _ = self .events .tell(Publish(ProposalApproved { @@ -572,3 +328,6 @@ impl ProposalManager { Ok(()) } } + +#[cfg(test)] +mod tests; diff --git a/server/crates/arbiter-server/src/actors/proposal_manager/store.rs b/server/crates/arbiter-server/src/actors/proposal_manager/store.rs new file mode 100644 index 0000000..f735a7e --- /dev/null +++ b/server/crates/arbiter-server/src/actors/proposal_manager/store.rs @@ -0,0 +1,409 @@ +//! Database access for [`super::ProposalManager`], behind a trait. +//! +//! The actor holds a `dyn ProposalStore` rather than a connection pool, so the quorum +//! rules can be exercised against a mock instead of a live SQLite file. + +use super::{Error, ProposalSummary, WAKEUP_DELAY_SECS}; +use crate::db::{ + self, + functions::unixepoch, + models::{ + NewProposal, NewProposalVote, NewRecoveryProposalVote, NewRecoveryWakeupRequest, + OperatorIdentityId, Proposal, ProposalId, ProposalStatus, RecoveryOperatorIdentityId, + SqliteTimestamp, + }, + proposal::{ProposalKind, ProposalKindTag}, + schema, +}; + +use async_trait::async_trait; +use chrono::Utc; +use diesel::{ + ExpressionMethods as _, QueryDsl, + dsl::{exists, select}, +}; +use diesel_async::{AsyncConnection as _, RunQueryDsl}; +use std::collections::HashMap; +use strum::IntoDiscriminant as _; + +/// Everything the quorum rules need to know about one proposal's votes. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Tally { + pub approve: i64, + pub reject: i64, + pub total_ordinary: i64, + pub total_recovery: i64, +} + +#[cfg_attr(test, mockall::automock)] +#[async_trait] +pub trait ProposalStore: Send + Sync + 'static { + /// Writes the proposal and its kind-specific rows in one transaction. + async fn create( + &self, + kind: ProposalKind, + initiator_id: OperatorIdentityId, + expires_at: SqliteTimestamp, + ) -> Result; + + async fn load(&self, id: ProposalId) -> Result; + + async fn load_kind(&self, id: ProposalId, tag: ProposalKindTag) -> Result; + + async fn has_voted( + &self, + id: ProposalId, + operator_id: OperatorIdentityId, + ) -> Result; + + async fn has_recovery_voted( + &self, + id: ProposalId, + recovery_operator_id: RecoveryOperatorIdentityId, + ) -> Result; + + async fn operator_public_key(&self, id: OperatorIdentityId) -> Result, Error>; + + async fn recovery_operator_public_key( + &self, + id: RecoveryOperatorIdentityId, + ) -> Result, Error>; + + async fn record_vote(&self, vote: NewProposalVote) -> Result<(), Error>; + + async fn record_recovery_vote(&self, vote: NewRecoveryProposalVote) -> Result<(), Error>; + + /// Vote counts for one proposal, alongside the size of each electorate. + async fn tally(&self, id: ProposalId) -> Result; + + async fn set_status(&self, id: ProposalId, status: ProposalStatus) -> Result<(), Error>; + + /// Pending, unexpired proposals this operator has not voted on yet. + async fn pending_for( + &self, + operator_id: OperatorIdentityId, + ) -> Result, Error>; + + /// True once an uncancelled wake-up request has outlived the dispute window. + async fn is_recovery_active(&self) -> Result; + + /// True while any wake-up request stands, whether or not the window has elapsed. + async fn has_uncancelled_wakeup(&self) -> Result; + + async fn request_wakeup(&self, operator_id: OperatorIdentityId) -> Result<(), Error>; + + /// Returns false when there was no uncancelled request to cancel. + async fn cancel_wakeup(&self, operator_id: OperatorIdentityId) -> Result; +} + +pub struct DieselProposalStore { + db: db::DatabasePool, +} + +impl DieselProposalStore { + pub const fn new(db: db::DatabasePool) -> Self { + Self { db } + } +} + +/// `NotFound` means the row is absent, which every caller reports as its own error. +fn missing(absent: Error) -> impl FnOnce(diesel::result::Error) -> Error { + move |e| match e { + diesel::result::Error::NotFound => absent, + other => Error::DatabaseQuery(other), + } +} + +#[async_trait] +impl ProposalStore for DieselProposalStore { + async fn create( + &self, + kind: ProposalKind, + initiator_id: OperatorIdentityId, + expires_at: SqliteTimestamp, + ) -> Result { + let id = self + .db + .get() + .await? + .transaction(async |conn| { + let id: ProposalId = diesel::insert_into(schema::proposal::table) + .values(&NewProposal { + kind: kind.discriminant(), + initiator_id, + expires_at, + }) + .returning(schema::proposal::id) + .get_result(conn) + .await?; + db::proposal::insert_kind(conn, id, &kind).await?; + Ok::<_, diesel::result::Error>(id) + }) + .await?; + + Ok(id) + } + + async fn load(&self, id: ProposalId) -> Result { + let mut conn = self.db.get().await?; + schema::proposal::table + .find(id) + .first(&mut conn) + .await + .map_err(missing(Error::ProposalNotFound)) + } + + async fn load_kind(&self, id: ProposalId, tag: ProposalKindTag) -> Result { + let mut conn = self.db.get().await?; + db::proposal::load_kind(&mut conn, id, tag) + .await + .map_err(Error::from) + } + + async fn has_voted( + &self, + id: ProposalId, + operator_id: OperatorIdentityId, + ) -> Result { + let mut conn = self.db.get().await?; + select(exists( + schema::proposal_vote::table + .filter(schema::proposal_vote::proposal_id.eq(id)) + .filter(schema::proposal_vote::operator_id.eq(operator_id)), + )) + .get_result(&mut conn) + .await + .map_err(Error::from) + } + + async fn has_recovery_voted( + &self, + id: ProposalId, + recovery_operator_id: RecoveryOperatorIdentityId, + ) -> Result { + let mut conn = self.db.get().await?; + select(exists( + schema::recovery_proposal_vote::table + .filter(schema::recovery_proposal_vote::proposal_id.eq(id)) + .filter( + schema::recovery_proposal_vote::recovery_operator_id.eq(recovery_operator_id), + ), + )) + .get_result(&mut conn) + .await + .map_err(Error::from) + } + + async fn operator_public_key(&self, id: OperatorIdentityId) -> Result, Error> { + let mut conn = self.db.get().await?; + schema::operator_identity::table + .find(id) + .select(schema::operator_identity::public_key) + .first(&mut conn) + .await + .map_err(missing(Error::OperatorNotFound)) + } + + async fn recovery_operator_public_key( + &self, + id: RecoveryOperatorIdentityId, + ) -> Result, Error> { + let mut conn = self.db.get().await?; + schema::recovery_operator_identity::table + .find(id) + .select(schema::recovery_operator_identity::public_key) + .first(&mut conn) + .await + .map_err(missing(Error::OperatorNotFound)) + } + + async fn record_vote(&self, vote: NewProposalVote) -> Result<(), Error> { + let mut conn = self.db.get().await?; + diesel::insert_into(schema::proposal_vote::table) + .values(&vote) + .execute(&mut conn) + .await?; + Ok(()) + } + + async fn record_recovery_vote(&self, vote: NewRecoveryProposalVote) -> Result<(), Error> { + let mut conn = self.db.get().await?; + diesel::insert_into(schema::recovery_proposal_vote::table) + .values(&vote) + .execute(&mut conn) + .await?; + Ok(()) + } + + async fn tally(&self, id: ProposalId) -> Result { + let mut conn = self.db.get().await?; + + let ordinary_approve: i64 = schema::proposal_vote::table + .filter(schema::proposal_vote::proposal_id.eq(id)) + .filter(schema::proposal_vote::approve.eq(true)) + .count() + .get_result(&mut conn) + .await?; + let recovery_approve: i64 = schema::recovery_proposal_vote::table + .filter(schema::recovery_proposal_vote::proposal_id.eq(id)) + .filter(schema::recovery_proposal_vote::approve.eq(true)) + .count() + .get_result(&mut conn) + .await?; + + let ordinary_reject: i64 = schema::proposal_vote::table + .filter(schema::proposal_vote::proposal_id.eq(id)) + .filter(schema::proposal_vote::approve.eq(false)) + .count() + .get_result(&mut conn) + .await?; + let recovery_reject: i64 = schema::recovery_proposal_vote::table + .filter(schema::recovery_proposal_vote::proposal_id.eq(id)) + .filter(schema::recovery_proposal_vote::approve.eq(false)) + .count() + .get_result(&mut conn) + .await?; + + let total_ordinary: i64 = schema::operator_identity::table + .count() + .get_result(&mut conn) + .await?; + let total_recovery: i64 = schema::recovery_operator_identity::table + .count() + .get_result(&mut conn) + .await?; + + Ok(Tally { + approve: ordinary_approve + recovery_approve, + reject: ordinary_reject + recovery_reject, + total_ordinary, + total_recovery, + }) + } + + async fn set_status(&self, id: ProposalId, status: ProposalStatus) -> Result<(), Error> { + let mut conn = self.db.get().await?; + diesel::update(schema::proposal::table.find(id)) + .set(schema::proposal::status.eq(status)) + .execute(&mut conn) + .await?; + Ok(()) + } + + async fn pending_for( + &self, + operator_id: OperatorIdentityId, + ) -> Result, Error> { + #[expect( + clippy::cast_possible_truncation, + clippy::as_conversions, + reason = "fixme! #84; this will break in 2038" + )] + let now_ts = Utc::now().timestamp() as i32; + + let mut conn = self.db.get().await?; + + let voted_ids: Vec = schema::proposal_vote::table + .filter(schema::proposal_vote::operator_id.eq(operator_id)) + .select(schema::proposal_vote::proposal_id) + .load(&mut conn) + .await?; + + let proposals: Vec = schema::proposal::table + .filter(schema::proposal::status.eq(ProposalStatus::Pending)) + .filter(schema::proposal::expires_at.gt(now_ts)) + .filter(diesel::dsl::not(schema::proposal::id.eq_any(&voted_ids))) + .load(&mut conn) + .await?; + + let ids: Vec = proposals.iter().map(|p| p.id).collect(); + let tallies: Vec<(ProposalId, bool, i64)> = schema::proposal_vote::table + .filter(schema::proposal_vote::proposal_id.eq_any(&ids)) + .group_by(( + schema::proposal_vote::proposal_id, + schema::proposal_vote::approve, + )) + .select(( + schema::proposal_vote::proposal_id, + schema::proposal_vote::approve, + diesel::dsl::count_star(), + )) + .load(&mut conn) + .await?; + + let mut by_proposal: HashMap = HashMap::new(); + for (proposal_id, approve, count) in tallies { + let entry = by_proposal.entry(proposal_id).or_insert((0, 0)); + if approve { + entry.0 += count; + } else { + entry.1 += count; + } + } + + Ok(proposals + .into_iter() + .map(|p| { + let (approve_count, reject_count) = + by_proposal.get(&p.id).copied().unwrap_or((0, 0)); + ProposalSummary { + id: p.id, + kind: p.kind, + initiator_id: p.initiator_id, + expires_at: p.expires_at, + approve_count, + reject_count, + } + }) + .collect()) + } + + async fn is_recovery_active(&self) -> Result { + let mut conn = self.db.get().await?; + select(exists( + schema::recovery_wakeup_request::table + .filter(schema::recovery_wakeup_request::cancelled_at.is_null()) + .filter( + schema::recovery_wakeup_request::requested_at + .le(unixepoch("now") - WAKEUP_DELAY_SECS), + ), + )) + .get_result(&mut conn) + .await + .map_err(Error::from) + } + + async fn has_uncancelled_wakeup(&self) -> Result { + let mut conn = self.db.get().await?; + select(exists(schema::recovery_wakeup_request::table.filter( + schema::recovery_wakeup_request::cancelled_at.is_null(), + ))) + .get_result(&mut conn) + .await + .map_err(Error::from) + } + + async fn request_wakeup(&self, operator_id: OperatorIdentityId) -> Result<(), Error> { + let mut conn = self.db.get().await?; + diesel::insert_into(schema::recovery_wakeup_request::table) + .values(&NewRecoveryWakeupRequest { + requested_by: operator_id, + }) + .execute(&mut conn) + .await?; + Ok(()) + } + + async fn cancel_wakeup(&self, operator_id: OperatorIdentityId) -> Result { + let mut conn = self.db.get().await?; + let rows = diesel::update(schema::recovery_wakeup_request::table) + .filter(schema::recovery_wakeup_request::cancelled_at.is_null()) + .set(( + schema::recovery_wakeup_request::cancelled_by.eq(Some(operator_id)), + schema::recovery_wakeup_request::cancelled_at.eq(Some(SqliteTimestamp::now())), + )) + .execute(&mut conn) + .await?; + Ok(rows > 0) + } +} diff --git a/server/crates/arbiter-server/src/actors/proposal_manager/tests.rs b/server/crates/arbiter-server/src/actors/proposal_manager/tests.rs new file mode 100644 index 0000000..7a8a607 --- /dev/null +++ b/server/crates/arbiter-server/src/actors/proposal_manager/tests.rs @@ -0,0 +1,222 @@ +//! The quorum rules, exercised without a database. +//! +//! These assertions are the point of [`super::store::ProposalStore`]: until the actor took +//! its data through a trait, checking that two of three operators carry an ordinary +//! proposal meant opening SQLite and registering operators first. + +use super::{ + ProposalManager, VoteOutcome, + store::{MockProposalStore, Tally}, +}; +use crate::{ + actors::GlobalActors, + crypto::governance::vote_message, + db::{ + models::{OperatorIdentityId, Proposal, ProposalId, ProposalStatus, SqliteTimestamp}, + proposal::ProposalKindTag, + }, +}; +use arbiter_crypto::authn::{SigningContext, SigningKey}; +use chrono::{Duration, Utc}; +use std::sync::Arc; + +const fn tally(approve: i64, reject: i64, ordinary: i64, recovery: i64) -> Tally { + Tally { + approve, + reject, + total_ordinary: ordinary, + total_recovery: recovery, + } +} + +#[test] +fn simple_majority_approves_at_two_of_three() { + assert_eq!( + ProposalManager::evaluate_quorum(&tally(2, 0, 3, 0), false), + VoteOutcome::Approved + ); +} + +#[test] +fn one_of_three_is_not_yet_a_majority() { + assert_eq!( + ProposalManager::evaluate_quorum(&tally(1, 0, 3, 0), false), + VoteOutcome::Pending + ); +} + +#[test] +fn full_quorum_kind_needs_every_voter() { + assert_eq!( + ProposalManager::evaluate_quorum(&tally(2, 0, 3, 0), true), + VoteOutcome::Pending, + "two of three must not carry a key-rotation proposal" + ); +} + +#[test] +fn recovery_voters_count_towards_full_quorum() { + assert_eq!( + ProposalManager::evaluate_quorum(&tally(3, 0, 2, 1), true), + VoteOutcome::Approved + ); + assert_eq!( + ProposalManager::evaluate_quorum(&tally(2, 0, 2, 1), true), + VoteOutcome::Pending, + "the sleeping recovery operator still owes a vote" + ); +} + +#[test] +fn rejection_is_decided_once_approval_is_unreachable() { + // Threshold is 2 of 3, so two rejections leave at most one approval available. + assert_eq!( + ProposalManager::evaluate_quorum(&tally(0, 2, 3, 0), false), + VoteOutcome::Rejected + ); + assert_eq!( + ProposalManager::evaluate_quorum(&tally(0, 1, 3, 0), false), + VoteOutcome::Pending, + "one rejection still leaves two approvals reachable" + ); +} + +#[test] +fn a_single_rejection_sinks_a_full_quorum_proposal() { + assert_eq!( + ProposalManager::evaluate_quorum(&tally(2, 1, 3, 0), true), + VoteOutcome::Rejected + ); +} + +fn pending_proposal(id: ProposalId, kind: ProposalKindTag) -> Proposal { + let now = Utc::now(); + Proposal { + id, + kind, + initiator_id: OperatorIdentityId::from_raw(1), + created_at: SqliteTimestamp::from(now), + expires_at: SqliteTimestamp::from(now + Duration::days(1)), + status: ProposalStatus::Pending, + } +} + +/// The mock earns its keep here: reaching quorum must flip the stored status to +/// `Approved` exactly once. Signature verification stays real -- only the database is +/// stubbed out. +#[tokio::test] +async fn reaching_quorum_marks_the_proposal_approved() { + let id = ProposalId::from_raw(1); + let voter = OperatorIdentityId::from_raw(1); + let key = SigningKey::generate(); + let signature = key + .sign_message(&vote_message(id, true), SigningContext::GovernanceVote) + .expect("signing a vote must succeed"); + let public_key = key.public_key().to_bytes(); + + let mut store = MockProposalStore::new(); + store + .expect_load() + .returning(move |id| Ok(pending_proposal(id, ProposalKindTag::TriggerRekey))); + store.expect_has_voted().returning(|_, _| Ok(false)); + store + .expect_operator_public_key() + .returning(move |_| Ok(public_key.clone())); + store.expect_record_vote().returning(|_| Ok(())); + store.expect_is_recovery_active().returning(|| Ok(false)); + store.expect_tally().returning(|_| Ok(tally(1, 0, 1, 0))); + store + .expect_set_status() + .withf(move |got, status| *got == id && *status == ProposalStatus::Approved) + .times(1) + .returning(|_, _| Ok(())); + store + .expect_load_kind() + .returning(|_, _| Ok(crate::db::proposal::ProposalKind::TriggerRekey)); + + let mut manager = + ProposalManager::with_store(Arc::new(store), GlobalActors::spawn_message_bus()); + + let outcome = manager + .cast_vote(id, voter, true, signature.to_bytes()) + .await + .expect("a valid vote must be accepted"); + + assert_eq!(outcome, VoteOutcome::Approved); +} + +/// A vote that does not reach the threshold must leave the stored status alone. +#[tokio::test] +async fn a_vote_short_of_quorum_does_not_touch_the_status() { + let id = ProposalId::from_raw(7); + let voter = OperatorIdentityId::from_raw(2); + let key = SigningKey::generate(); + let signature = key + .sign_message(&vote_message(id, true), SigningContext::GovernanceVote) + .expect("signing a vote must succeed"); + let public_key = key.public_key().to_bytes(); + + let mut store = MockProposalStore::new(); + store + .expect_load() + .returning(move |id| Ok(pending_proposal(id, ProposalKindTag::ApproveSdkClient))); + store.expect_has_voted().returning(|_, _| Ok(false)); + store + .expect_operator_public_key() + .returning(move |_| Ok(public_key.clone())); + store.expect_record_vote().returning(|_| Ok(())); + store.expect_is_recovery_active().returning(|| Ok(false)); + store.expect_tally().returning(|_| Ok(tally(1, 0, 3, 0))); + store.expect_set_status().never(); + + let mut manager = + ProposalManager::with_store(Arc::new(store), GlobalActors::spawn_message_bus()); + + let outcome = manager + .cast_vote(id, voter, true, signature.to_bytes()) + .await + .expect("a valid vote must be accepted"); + + assert_eq!(outcome, VoteOutcome::Pending); +} + +/// A sleeping recovery electorate must not raise the bar for an ordinary proposal. +#[tokio::test] +async fn sleeping_recovery_operators_do_not_count_towards_quorum() { + let id = ProposalId::from_raw(9); + let voter = OperatorIdentityId::from_raw(3); + let key = SigningKey::generate(); + let signature = key + .sign_message(&vote_message(id, true), SigningContext::GovernanceVote) + .expect("signing a vote must succeed"); + let public_key = key.public_key().to_bytes(); + + let mut store = MockProposalStore::new(); + store + .expect_load() + .returning(move |id| Ok(pending_proposal(id, ProposalKindTag::ApproveSdkClient))); + store.expect_has_voted().returning(|_, _| Ok(false)); + store + .expect_operator_public_key() + .returning(move |_| Ok(public_key.clone())); + store.expect_record_vote().returning(|_| Ok(())); + store.expect_is_recovery_active().returning(|| Ok(false)); + // Two recovery operators exist but are asleep, so the threshold stays at 1 of 1. + store.expect_tally().returning(|_| Ok(tally(1, 0, 1, 2))); + store.expect_set_status().times(1).returning(|_, _| Ok(())); + store.expect_load_kind().returning(|_, _| { + Ok(crate::db::proposal::ProposalKind::ApproveSdkClient( + crate::db::proposal::approve_sdk_client::Settings { client_id: 1 }, + )) + }); + + let mut manager = + ProposalManager::with_store(Arc::new(store), GlobalActors::spawn_message_bus()); + + let outcome = manager + .cast_vote(id, voter, true, signature.to_bytes()) + .await + .expect("a valid vote must be accepted"); + + assert_eq!(outcome, VoteOutcome::Approved); +} diff --git a/server/crates/arbiter-server/tests/governance.rs b/server/crates/arbiter-server/tests/governance.rs index 1819809..2c5ba2d 100644 --- a/server/crates/arbiter-server/tests/governance.rs +++ b/server/crates/arbiter-server/tests/governance.rs @@ -26,7 +26,6 @@ use arbiter_server::db::schema::{ }; use diesel::{ExpressionMethods, QueryDsl, insert_into}; use diesel_async::RunQueryDsl; -use std::future::Future; /// Retries `probe` until it yields a value, then returns it. ///