perf(proposal): replace the per-proposal tally loop with one grouped query

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

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;