refactor(proposal): put DB access behind a mockable ProposalStore trait

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"

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

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,
+            NewProposalVote, NewRecoveryProposalVote, OperatorIdentityId, Proposal, ProposalId,
-            OperatorIdentityId, Proposal, ProposalId, ProposalStatus, RecoveryOperatorIdentityId,
+            ProposalStatus, RecoveryOperatorIdentityId, SqliteTimestamp,
-            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 std::sync::Arc;
-use strum::IntoDiscriminant as _;
 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<dyn ProposalStore>,
     pub(crate) events: ActorRef<MessageBus>,
 }
 
 impl ProposalManager {
-    pub const fn new(db: db::DatabasePool, events: ActorRef<MessageBus>) -> Self {
+    pub fn new(db: db::DatabasePool, events: ActorRef<MessageBus>) -> Self {
-        Self { db, events }
+        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<dyn ProposalStore>,
+        events: ActorRef<MessageBus>,
+    ) -> 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
+        self.store.create(kind, initiator_id, expires_at).await
-            .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)
     }
 
     #[message]
     pub async fn query_pending(&mut self, operator_id: OperatorIdentityId) -> Vec<ProposalSummary> {
-        #[expect(
+        self.store
-            clippy::cast_possible_truncation,
+            .pending_for(operator_id)
-            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<ProposalId> = schema::proposal_vote::table
-            .filter(schema::proposal_vote::operator_id.eq(operator_id))
-            .select(schema::proposal_vote::proposal_id)
-            .load(&mut conn)
             .await
-            .unwrap_or_default();
+            .unwrap_or_else(|e| {
-
+                warn!(?e, "query_pending failed");
-        let proposals: Vec<Proposal> = schema::proposal::table
+                vec![]
-            .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<ProposalId> = 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<ProposalId, (i64, i64)> = 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,
-                }
             })
-            .collect()
     }
 
     #[message]
@@ -209,141 +136,35 @@ impl ProposalManager {
         approve: bool,
         signature: Vec<u8>,
     ) -> Result<VoteOutcome, Error> {
-        let mut conn = self.db.get().await?;
+        let proposal = self.store.load(proposal_id).await?;
 
-        // Load proposal — must exist
+        // Checked before the status check so AlreadyVoted takes priority.
-        let proposal: Proposal = schema::proposal::table
+        if self.store.has_voted(proposal_id, operator_id).await? {
-            .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 {
             return Err(Error::AlreadyVoted);
         }
 
-        if proposal.status != ProposalStatus::Pending {
+        Self::check_votable(&proposal)?;
-            return Err(Error::ProposalNotPending);
-        }
 
-        if proposal.expires_at.0 <= Utc::now() {
+        let public_key = self.store.operator_public_key(operator_id).await?;
-            return Err(Error::ProposalExpired);
+        governance::verify_vote(&public_key, proposal_id, approve, &signature)
-        }
-
-        // Load operator public key from operator_identity
-        let pubkey_bytes: Vec<u8> = 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)
             .map_err(|_| Error::InvalidSignature)?;
 
-        // Insert vote
+        self.store
-        diesel::insert_into(schema::proposal_vote::table)
+            .record_vote(NewProposalVote {
-            .values(&NewProposalVote {
                 proposal_id,
                 operator_id,
                 approve,
                 signature,
             })
-            .execute(&mut conn)
             .await?;
 
-        // Quorum check
+        let mut tally = self.store.tally(proposal_id).await?;
-        let total_operators: i64 = schema::operator_identity::table
+        // §3.5: recovery operators only join the electorate once they are awake.
-            .count()
+        if !self.store.is_recovery_active().await? {
-            .get_result(&mut conn)
+            tally.total_recovery = 0;
-            .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 total_eligible = total_operators + total_recovery;
+        self.settle(&proposal, &tally).await
-        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)
     }
 
     /// §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.store.has_uncancelled_wakeup().await? {
-        if Self::has_uncancelled_wakeup(&mut conn).await? {
             return Err(Error::WakeupAlreadyPending);
         }
-        diesel::insert_into(schema::recovery_wakeup_request::table)
+        self.store.request_wakeup(operator_id).await
-            .values(&NewRecoveryWakeupRequest {
-                requested_by: operator_id,
-            })
-            .execute(&mut conn)
-            .await?;
-        Ok(())
     }
 
     /// §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?;
+        if self.store.cancel_wakeup(operator_id).await? {
-        let rows_updated = diesel::update(schema::recovery_wakeup_request::table)
+            Ok(())
-            .filter(schema::recovery_wakeup_request::cancelled_at.is_null())
+        } else {
-            .set((
+            Err(Error::NoActiveWakeup)
-                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);
         }
-        Ok(())
     }
 
     /// §3.5: Recovery operators may only vote on operator replacement proposals.
@@ -398,151 +204,106 @@ impl ProposalManager {
         approve: bool,
         signature: Vec<u8>,
     ) -> Result<VoteOutcome, Error> {
-        let mut conn = self.db.get().await?;
+        let proposal = self.store.load(proposal_id).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),
-            })?;
 
         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(
+        if self
-            schema::recovery_proposal_vote::table
+            .store
-                .filter(schema::recovery_proposal_vote::proposal_id.eq(proposal_id))
+            .has_recovery_voted(proposal_id, recovery_operator_id)
-                .filter(
+            .await?
-                    schema::recovery_proposal_vote::recovery_operator_id.eq(recovery_operator_id),
+        {
-                ),
-        ))
-        .get_result(&mut conn)
-        .await?;
-        if already_voted {
             return Err(Error::AlreadyVoted);
         }
 
-        if proposal.status != ProposalStatus::Pending {
+        Self::check_votable(&proposal)?;
-            return Err(Error::ProposalNotPending);
-        }
 
-        if proposal.expires_at.0 <= Utc::now() {
+        let public_key = self
-            return Err(Error::ProposalExpired);
+            .store
-        }
+            .recovery_operator_public_key(recovery_operator_id)
-
+            .await?;
-        let pubkey_bytes: Vec<u8> = schema::recovery_operator_identity::table
+        governance::verify_vote(&public_key, proposal_id, approve, &signature)
-            .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)
             .map_err(|_| Error::InvalidSignature)?;
 
-        diesel::insert_into(schema::recovery_proposal_vote::table)
+        self.store
-            .values(&NewRecoveryProposalVote {
+            .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 tally = self.store.tally(proposal_id).await?;
-        let total_ordinary: i64 = schema::operator_identity::table
+        self.settle(&proposal, &tally).await
-            .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)
     }
 }
 
 impl ProposalManager {
-    const WAKEUP_DELAY_SECS: i32 = 14 * 24 * 60 * 60;
+    /// A vote only counts while the proposal is still open.
-
+    fn check_votable(proposal: &Proposal) -> Result<(), Error> {
-    /// Returns true when an uncancelled wakeup request has passed the 14-day dispute window.
+        if proposal.status != ProposalStatus::Pending {
-    async fn is_recovery_active_conn(conn: &mut db::DatabaseConnection) -> Result<bool, Error> {
+            return Err(Error::ProposalNotPending);
-        select(exists(
+        }
-            schema::recovery_wakeup_request::table
+        if proposal.expires_at.0 <= Utc::now() {
-                .filter(schema::recovery_wakeup_request::cancelled_at.is_null())
+            return Err(Error::ProposalExpired);
-                .filter(
+        }
-                    schema::recovery_wakeup_request::requested_at
+        Ok(())
-                        .le(unixepoch("now") - Self::WAKEUP_DELAY_SECS),
-                ),
-        ))
-        .get_result(conn)
-        .await
-        .map_err(Error::from)
     }
 
-    /// Returns true when there is any uncancelled wakeup request (pending or active).
+    /// Pure quorum arithmetic — no I/O, so the rules can be tested directly (§3.3).
-    async fn has_uncancelled_wakeup(conn: &mut db::DatabaseConnection) -> Result<bool, Error> {
+    ///
-        select(exists(schema::recovery_wakeup_request::table.filter(
+    /// A proposal is rejected once approval has become unreachable: even if every voter
-            schema::recovery_wakeup_request::cancelled_at.is_null(),
+    /// who has not spoken yet approved, the threshold could not be met.
-        )))
+    #[must_use]
-        .get_result(conn)
+    pub(crate) const fn evaluate_quorum(tally: &Tally, requires_full_quorum: bool) -> VoteOutcome {
-        .await
+        let total_eligible = tally.total_ordinary + tally.total_recovery;
-        .map_err(Error::from)
+
+        #[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<VoteOutcome, Error> {
+        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(
+    async fn announce_approval(&self, proposal: &Proposal) -> Result<(), Error> {
-        &self,
+        self.store
-        conn: &mut db::DatabaseConnection,
+            .set_status(proposal.id, ProposalStatus::Approved)
-        proposal: &Proposal,
-    ) -> Result<(), Error> {
-        diesel::update(schema::proposal::table.find(proposal.id))
-            .set(schema::proposal::status.eq(ProposalStatus::Approved))
-            .execute(conn)
             .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;

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<ProposalId, Error>;
+
+    async fn load(&self, id: ProposalId) -> Result<Proposal, Error>;
+
+    async fn load_kind(&self, id: ProposalId, tag: ProposalKindTag) -> Result<ProposalKind, Error>;
+
+    async fn has_voted(
+        &self,
+        id: ProposalId,
+        operator_id: OperatorIdentityId,
+    ) -> Result<bool, Error>;
+
+    async fn has_recovery_voted(
+        &self,
+        id: ProposalId,
+        recovery_operator_id: RecoveryOperatorIdentityId,
+    ) -> Result<bool, Error>;
+
+    async fn operator_public_key(&self, id: OperatorIdentityId) -> Result<Vec<u8>, Error>;
+
+    async fn recovery_operator_public_key(
+        &self,
+        id: RecoveryOperatorIdentityId,
+    ) -> Result<Vec<u8>, 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<Tally, Error>;
+
+    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<Vec<ProposalSummary>, Error>;
+
+    /// True once an uncancelled wake-up request has outlived the dispute window.
+    async fn is_recovery_active(&self) -> Result<bool, Error>;
+
+    /// True while any wake-up request stands, whether or not the window has elapsed.
+    async fn has_uncancelled_wakeup(&self) -> Result<bool, Error>;
+
+    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<bool, Error>;
+}
+
+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<ProposalId, Error> {
+        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<Proposal, Error> {
+        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<ProposalKind, Error> {
+        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<bool, Error> {
+        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<bool, Error> {
+        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<Vec<u8>, 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<Vec<u8>, 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<Tally, Error> {
+        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<Vec<ProposalSummary>, 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<ProposalId> = 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<Proposal> = 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<ProposalId> = 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<ProposalId, (i64, i64)> = 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<bool, Error> {
+        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<bool, Error> {
+        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<bool, Error> {
+        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)
+    }
+}

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);
+}

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.
 ///