From e9496da78c4981b4bb9907d908b8cd1f836a8bdf Mon Sep 17 00:00:00 2001 From: CleverWild Date: Thu, 27 Aug 2026 14:00:14 +0200 Subject: [PATCH] perf(proposal): replace the per-proposal tally loop with one grouped query --- .../src/actors/proposal_manager.rs | 66 ++++++++----- .../crates/arbiter-server/tests/governance.rs | 92 +++++++++++++++++++ 2 files changed, 133 insertions(+), 25 deletions(-) diff --git a/server/crates/arbiter-server/src/actors/proposal_manager.rs b/server/crates/arbiter-server/src/actors/proposal_manager.rs index 8f59928..7945bf1 100644 --- a/server/crates/arbiter-server/src/actors/proposal_manager.rs +++ b/server/crates/arbiter-server/src/actors/proposal_manager.rs @@ -22,6 +22,7 @@ use diesel::{ }; use diesel_async::{AsyncConnection as _, RunQueryDsl}; use kameo::{Actor, actor::ActorRef, messages}; +use std::collections::HashMap; use strum::IntoDiscriminant as _; use tracing::{error, warn}; @@ -168,32 +169,47 @@ impl ProposalManager { .await .unwrap_or_default(); - let mut summaries = Vec::with_capacity(proposals.len()); - for p in proposals { - let approve_count: i64 = schema::proposal_vote::table - .filter(schema::proposal_vote::proposal_id.eq(p.id)) - .filter(schema::proposal_vote::approve.eq(true)) - .count() - .get_result(&mut conn) - .await - .unwrap_or(0); - let reject_count: i64 = schema::proposal_vote::table - .filter(schema::proposal_vote::proposal_id.eq(p.id)) - .filter(schema::proposal_vote::approve.eq(false)) - .count() - .get_result(&mut conn) - .await - .unwrap_or(0); - summaries.push(ProposalSummary { - id: p.id, - kind: p.kind, - initiator_id: p.initiator_id, - expires_at: p.expires_at, - approve_count, - reject_count, - }); + 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; + } } - summaries + + 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] diff --git a/server/crates/arbiter-server/tests/governance.rs b/server/crates/arbiter-server/tests/governance.rs index 55e2c51..226721a 100644 --- a/server/crates/arbiter-server/tests/governance.rs +++ b/server/crates/arbiter-server/tests/governance.rs @@ -374,6 +374,98 @@ async fn invalid_signature_rejected() { )); } +#[tokio::test] +async fn query_pending_reports_a_tally_per_proposal() { + let db = db::create_test_pool().await; + let actors = GlobalActors::spawn(db.clone()).await.unwrap(); + actors + .vault + .ask(Bootstrap { + seal_key: KeyCell::from([0u8; 32]), + }) + .await + .unwrap(); + + // Three operators, so one vote stays below the 2-of-3 threshold and every + // proposal is still pending when it is queried. + let approver = authn::SigningKey::generate(); + let rejecter = authn::SigningKey::generate(); + let watcher = authn::SigningKey::generate(); + let approver_id = register_operator(&db, &approver.public_key()).await; + let rejecter_id = register_operator(&db, &rejecter.public_key()).await; + let watcher_id = register_operator(&db, &watcher.public_key()).await; + + let cast = async |proposal_id, voter_id, key: &authn::SigningKey, approve| { + let sig = key + .sign_message( + &make_vote_message(proposal_id, approve), + SigningContext::GovernanceVote, + ) + .unwrap(); + actors + .proposal_manager + .ask(CastVote { + proposal_id, + operator_id: voter_id, + approve, + signature: sig.to_bytes(), + }) + .await + .unwrap() + }; + + let mut ids = Vec::new(); + for client_id in 1..=3 { + let id = actors + .proposal_manager + .ask(CreateProposal { + kind: ProposalKind::ApproveSdkClient(approve_sdk_client::Settings { client_id }), + initiator_id: watcher_id, + ttl_secs: None, + }) + .await + .unwrap(); + ids.push(id); + } + + // First proposal: one approval. Second: one rejection. Third: neither. + assert_eq!( + cast(ids[0], approver_id, &approver, true).await, + VoteOutcome::Pending + ); + assert_eq!( + cast(ids[1], rejecter_id, &rejecter, false).await, + VoteOutcome::Pending + ); + + let summaries = actors + .proposal_manager + .ask(QueryPending { + operator_id: watcher_id, + }) + .await + .unwrap(); + assert_eq!(summaries.len(), 3, "the watcher has voted on nothing"); + + for summary in summaries { + let (approve, reject) = match summary.id { + id if id == ids[0] => (1, 0), + id if id == ids[1] => (0, 1), + _ => (0, 0), + }; + assert_eq!( + summary.approve_count, approve, + "approvals of {:?}", + summary.id + ); + assert_eq!( + summary.reject_count, reject, + "rejections of {:?}", + summary.id + ); + } +} + #[tokio::test] async fn query_pending_excludes_already_voted() { let db = db::create_test_pool().await;