confium_coordinator/coordinator/
batch.rs

1//! Batch signing — coordinate multiple messages under one batch ID.
2//!
3//! High-volume signers produce many signatures per quorum activation.
4//! The batch API creates N sessions under one batch ID, amortizing
5//! quorum coordination overhead. Each message gets its own session
6//! (so each gets its own audit trail), but they share the same
7//! quorum, threshold, and lifecycle.
8
9use crate::coordinator::coordinator::Coordinator;
10use crate::coordinator::session::{
11    AggregatedSignature, SessionError, SessionId, SessionRequest, SessionState,
12};
13use std::collections::HashMap;
14
15/// A batch signing request: N messages, same quorum.
16#[derive(Debug, Clone)]
17pub struct BatchSessionRequest {
18    /// Quorum authorizing all signatures in this batch.
19    pub quorum_id: String,
20    /// Threshold scheme.
21    pub scheme: String,
22    /// Messages to sign (one session per message).
23    pub messages: Vec<Vec<u8>>,
24    /// Threshold T.
25    pub threshold: u32,
26    /// Total party count N.
27    pub num_parties: u32,
28    /// Unlock window in minutes.
29    pub unlock_window_minutes: u32,
30    /// Requesting actor.
31    pub requested_by: String,
32}
33
34/// A batch of related sessions tracked under one batch ID.
35#[derive(Debug, Clone)]
36pub struct BatchSession {
37    /// Batch identifier.
38    pub batch_id: String,
39    /// Individual session IDs (one per message).
40    pub session_ids: Vec<SessionId>,
41    /// The original batch request.
42    pub request: BatchSessionRequest,
43}
44
45/// Errors during batch operations.
46#[derive(Debug, thiserror::Error)]
47pub enum BatchError {
48    /// A session in the batch failed.
49    #[error("session {session_id} failed: {error}")]
50    SessionFailed {
51        /// Which session failed.
52        session_id: SessionId,
53        /// The error.
54        error: String,
55    },
56    /// Batch not found.
57    #[error("batch not found: {0}")]
58    NotFound(String),
59    /// Empty batch.
60    #[error("batch has no messages")]
61    Empty,
62    /// Session creation failed.
63    #[error("session creation failed: {0}")]
64    SessionCreation(#[from] SessionError),
65}
66
67/// Batch session manager. Wraps a Coordinator to provide batch
68/// creation and aggregation.
69pub struct BatchSigner<'a> {
70    coordinator: &'a mut Coordinator,
71    batches: HashMap<String, BatchSession>,
72    next_batch_id: u64,
73}
74
75impl<'a> BatchSigner<'a> {
76    /// Create a new batch signer wrapping a coordinator.
77    pub fn new(coordinator: &'a mut Coordinator) -> Self {
78        Self {
79            coordinator,
80            batches: HashMap::new(),
81            next_batch_id: 0,
82        }
83    }
84
85    /// Create a batch of sessions — one per message.
86    /// Returns the batch ID and the list of session IDs.
87    pub fn create_batch(&mut self, request: BatchSessionRequest) -> Result<String, BatchError> {
88        if request.messages.is_empty() {
89            return Err(BatchError::Empty);
90        }
91
92        let batch_id = format!("batch-{}", self.next_batch_id);
93        self.next_batch_id += 1;
94
95        let mut session_ids = Vec::with_capacity(request.messages.len());
96        for message in &request.messages {
97            let req = SessionRequest {
98                quorum_id: request.quorum_id.clone(),
99                scheme: request.scheme.clone(),
100                message: message.clone(),
101                threshold: request.threshold,
102                num_parties: request.num_parties,
103                unlock_window_minutes: request.unlock_window_minutes,
104                requested_by: request.requested_by.clone(),
105            };
106            let sid = self.coordinator.create_session(req)?;
107            session_ids.push(sid);
108        }
109
110        let count = session_ids.len();
111        self.batches.insert(
112            batch_id.clone(),
113            BatchSession {
114                batch_id: batch_id.clone(),
115                session_ids,
116                request,
117            },
118        );
119
120        tracing::info!(batch_id = %batch_id, sessions = count, "batch created");
121        Ok(batch_id)
122    }
123
124    /// Aggregate all sessions in a batch. All sessions must have
125    /// received T shares. Returns one signature per message.
126    pub fn aggregate_batch(
127        &mut self,
128        batch_id: &str,
129    ) -> Result<Vec<AggregatedSignature>, BatchError> {
130        let batch = self
131            .batches
132            .get(batch_id)
133            .ok_or_else(|| BatchError::NotFound(batch_id.into()))?;
134
135        let session_ids: Vec<SessionId> = batch.session_ids.clone();
136        let mut results = Vec::with_capacity(session_ids.len());
137
138        for sid in session_ids {
139            match self.coordinator.aggregate(&sid) {
140                Ok(sig) => results.push(sig),
141                Err(e) => {
142                    return Err(BatchError::SessionFailed {
143                        session_id: sid,
144                        error: format!("{e:?}"),
145                    });
146                }
147            }
148        }
149
150        tracing::info!(batch_id = %batch_id, signatures = results.len(), "batch aggregated");
151        Ok(results)
152    }
153
154    /// Get the session IDs for a batch.
155    pub fn batch_session_ids(&self, batch_id: &str) -> Option<&[SessionId]> {
156        self.batches.get(batch_id).map(|b| b.session_ids.as_slice())
157    }
158
159    /// Number of batches managed.
160    pub fn batch_count(&self) -> usize {
161        self.batches.len()
162    }
163
164    /// Check if all sessions in a batch have reached the threshold
165    /// share count (ready for aggregation).
166    pub fn is_batch_ready(&self, batch_id: &str) -> bool {
167        let batch = match self.batches.get(batch_id) {
168            Some(b) => b,
169            None => return false,
170        };
171        batch.session_ids.iter().all(|sid| {
172            if let (Some(threshold), Some(count)) = (
173                self.coordinator.session_threshold(sid),
174                self.coordinator.session_share_count(sid),
175            ) {
176                count >= threshold as usize
177            } else {
178                false
179            }
180        })
181    }
182
183    /// Get batch states as a summary.
184    pub fn batch_states(&self, batch_id: &str) -> Option<Vec<SessionState>> {
185        self.batches.get(batch_id).map(|batch| {
186            batch
187                .session_ids
188                .iter()
189                .filter_map(|sid| self.coordinator.session_state(sid))
190                .collect()
191        })
192    }
193}
194
195#[cfg(test)]
196mod tests {
197    use super::*;
198    use crate::coordinator::coordinator::Coordinator;
199
200    fn make_batch_request(n: usize) -> BatchSessionRequest {
201        BatchSessionRequest {
202            quorum_id: "q1".into(),
203            scheme: "CMP20".into(),
204            messages: (0..n).map(|i| vec![i as u8; 32]).collect(),
205            threshold: 2,
206            num_parties: 3,
207            unlock_window_minutes: 60,
208            requested_by: "tester".into(),
209        }
210    }
211
212    #[test]
213    fn create_batch_produces_n_sessions() {
214        let mut coord = Coordinator::new();
215        let mut batcher = BatchSigner::new(&mut coord);
216        let batch_id = batcher.create_batch(make_batch_request(5)).unwrap();
217        assert!(batch_id.starts_with("batch-"));
218        assert_eq!(batcher.batch_session_ids(&batch_id).unwrap().len(), 5);
219        assert_eq!(batcher.batch_count(), 1);
220        assert_eq!(coord.session_count(), 5);
221    }
222
223    #[test]
224    fn empty_batch_rejected() {
225        let mut coord = Coordinator::new();
226        let mut batcher = BatchSigner::new(&mut coord);
227        let req = BatchSessionRequest {
228            messages: vec![],
229            ..make_batch_request(0)
230        };
231        assert!(batcher.create_batch(req).is_err());
232    }
233
234    #[test]
235    fn multiple_batches_have_incrementing_ids() {
236        let mut coord = Coordinator::new();
237        let mut batcher = BatchSigner::new(&mut coord);
238        let id1 = batcher.create_batch(make_batch_request(1)).unwrap();
239        let id2 = batcher.create_batch(make_batch_request(1)).unwrap();
240        assert_ne!(id1, id2);
241    }
242
243    #[test]
244    fn aggregate_unknown_batch_errors() {
245        let mut coord = Coordinator::new();
246        let mut batcher = BatchSigner::new(&mut coord);
247        assert!(batcher.aggregate_batch("nonexistent").is_err());
248    }
249
250    #[test]
251    fn batch_ready_false_when_shares_missing() {
252        let mut coord = Coordinator::new();
253        let mut batcher = BatchSigner::new(&mut coord);
254        let batch_id = batcher.create_batch(make_batch_request(2)).unwrap();
255        assert!(!batcher.is_batch_ready(&batch_id));
256    }
257
258    #[test]
259    fn batch_states_returns_per_session_states() {
260        let mut coord = Coordinator::new();
261        let mut batcher = BatchSigner::new(&mut coord);
262        let batch_id = batcher.create_batch(make_batch_request(3)).unwrap();
263        let states = batcher.batch_states(&batch_id).unwrap();
264        assert_eq!(states.len(), 3);
265        assert!(states.iter().all(|s| *s == SessionState::Pending));
266    }
267
268    #[test]
269    fn single_message_batch_works() {
270        let mut coord = Coordinator::new();
271        let mut batcher = BatchSigner::new(&mut coord);
272        let batch_id = batcher.create_batch(make_batch_request(1)).unwrap();
273        assert_eq!(batcher.batch_session_ids(&batch_id).unwrap().len(), 1);
274    }
275}
confium_coordinator/coordinator/batch.rs

confium_coordinator/coordinator/
batch.rs
