16810179837

Committed 07 Aug 2025 04:27PM UTC coverage: 55.941% (+0.02%) from 55.922%

Build # 16810179837

Build Type

Pull #1845

github

Committed by

web-flow

Commit Message

Merge 02139528b into 24cadc364

Pull Request Pull Request #1845: cargo fmt

Run Details

20 of 24 new or added lines in 3 files covered. (83.33%)

13 existing lines in 2 files now uncovered.

3889 of 6952 relevant lines covered (55.94%)

60.24 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/cli/src/tasks.rs

use crate::{CliResult, DetermineAccountId, Error, Output};
use clap::Subcommand;
use divviup_client::{
    BigUint, DivviupClient, Histogram, NewTask, Ratio, SumVec, Uuid, Vdaf,
    dp_strategy::{self, PureDpBudget, PureDpDiscreteLaplace},
};
use humantime::{Duration, Timestamp};
use std::time::SystemTime;
use time::OffsetDateTime;

#[derive(clap::ValueEnum, Clone, Debug)]
pub enum VdafName {
    Count,
    Histogram,
    Sum,
    CountVec,
    SumVec,
}

#[derive(clap::ValueEnum, Clone, Debug)]
pub enum DpStrategy {
    PureDpDiscreteLaplace,
}

#[derive(Subcommand, Debug)]
pub enum TaskAction {
    /// list all tasks for the target account
    List,

    /// retrieve details of a single task. this also refreshes cached data, such as metrics.
    Get { task_id: String },

    /// create a new task for the target account
    Create {
        #[arg(long)]
        name: String,
        #[arg(long)]
        leader_aggregator_id: Uuid,
        #[arg(long)]
        helper_aggregator_id: Uuid,
        #[arg(long)]
        vdaf: VdafName,
        #[arg(long)]
        min_batch_size: u64,
        #[arg(long)]
        max_batch_size: Option<u64>,
        #[arg(long, requires = "max_batch_size")]
        batch_time_window_size: Option<Duration>,
        #[arg(long)]
        time_precision: Duration,
        #[arg(long)]
        collector_credential_id: Uuid,
        #[arg(long, value_delimiter = ',')]
        categorical_buckets: Option<Vec<String>>,
        #[arg(long, value_delimiter = ',')]
        continuous_buckets: Option<Vec<u64>>,
        #[arg(long, required_if_eq_any([("vdaf", "count_vec"), ("vdaf", "sum_vec")]))]
        length: Option<u64>,
        #[arg(long, required_if_eq_any([("vdaf", "sum"), ("vdaf", "sum_vec")]))]
        bits: Option<u8>,
        #[arg(long)]
        chunk_length: Option<u64>,
        #[arg(long, requires = "differential_privacy_epsilon")]
        differential_privacy_strategy: Option<DpStrategy>,
        #[arg(long, requires = "differential_privacy_strategy")]
        differential_privacy_epsilon: Option<f64>,
    },

    /// rename a task
    Rename { task_id: String, name: String },

    /// delete a task
    Delete {
        task_id: String,
        /// delete the task even if the aggregators are unreachable
        #[arg(long, action)]
        force: bool,
    },

    /// set the expiration date of a task
    SetExpiration {
        task_id: String,
        /// the date and time to set to.
        ///
        /// if omitted, unset the expiration. the format is RFC 3339.
        expiration: Option<Timestamp>,
        /// set the expiration to the current time, effectively disabling the task.
        #[arg(long, action, conflicts_with = "expiration")]
        now: bool,
    },
}

impl TaskAction {
    pub(crate) async fn run(
        self,
        account_id: DetermineAccountId,
        client: DivviupClient,
        output: Output,
    ) -> CliResult {
        let account_id = account_id.await?;

        match self {
            TaskAction::List => output.display(client.tasks(account_id).await?),
            TaskAction::Get { task_id } => output.display(client.task(&task_id).await?),
            TaskAction::Create {
                name,
                leader_aggregator_id,
                helper_aggregator_id,
                vdaf,
                min_batch_size,
                max_batch_size,
                batch_time_window_size,
                time_precision,
                collector_credential_id,
                categorical_buckets,
                continuous_buckets,
                length,
                bits,
                chunk_length,
                differential_privacy_strategy,
                differential_privacy_epsilon,
            } => {
                let vdaf = match vdaf {
                    VdafName::Count => {
                        if differential_privacy_strategy.is_some()
                            || differential_privacy_epsilon.is_some()
                        {
                            return Err(Error::Other(
                                "differential privacy noise is not yet supported with Prio3Count"
                                    .into(),
                            ));
                        }
                        Vdaf::Count
                    }
                    VdafName::Histogram => {
                        let dp_strategy =
                            match (differential_privacy_strategy, differential_privacy_epsilon) {
                                (None, None) => dp_strategy::Prio3Histogram::NoDifferentialPrivacy,
                                (None, Some(_)) => {
                                    return Err(Error::Other(
                                        "missing differential-privacy-strategy".into(),
                                    ));
                                }
                                (Some(_), None) => {
                                    return Err(Error::Other(
                                        "missing differential-privacy-epsilon".into(),
                                    ));
                                }
                                (Some(DpStrategy::PureDpDiscreteLaplace), Some(epsilon)) => {
                                    dp_strategy::Prio3Histogram::PureDpDiscreteLaplace(
                                        PureDpDiscreteLaplace {
                                            budget: PureDpBudget {
                                                epsilon: float_to_biguint_ratio(epsilon)
                                                    .ok_or_else(|| {
                                                        Error::Other("invalid epsilon".into())
                                                    })?,
                                            },
                                        },
                                    )
                                }
                            };
                        match (length, categorical_buckets, continuous_buckets) {
                            (Some(length), None, None) => Vdaf::Histogram(Histogram::Length {
                                length,
                                chunk_length,
                                dp_strategy,
                            }),
                            (None, Some(buckets), None) => {
                                Vdaf::Histogram(Histogram::Categorical {
                                    buckets,
                                    chunk_length,
                                    dp_strategy,
                                })
                            }
                            (None, None, Some(buckets)) => Vdaf::Histogram(Histogram::Continuous {
                                buckets,
                                chunk_length,
                                dp_strategy,
                            }),
                            (None, None, None) => {
                                return Err(Error::Other("continuous-buckets, categorical-buckets, or length are required for histogram vdaf".into()));
                            }
                            _ => {
                                return Err(Error::Other("continuous-buckets, categorical-buckets, and length are mutually exclusive".into()));
                            }
                        }
                    }
                    VdafName::Sum => {
                        if differential_privacy_strategy.is_some()
                            || differential_privacy_epsilon.is_some()
                        {
                            return Err(Error::Other(
                                "differential privacy noise is not yet supported with Prio3Sum"
                                    .into(),
                            ));
                        }
                        Vdaf::Sum {
                            bits: bits.unwrap(),
                        }
                    }
                    VdafName::CountVec => {
                        if differential_privacy_strategy.is_some()
                            || differential_privacy_epsilon.is_some()
                        {
                            return Err(Error::Other(
                                "differential privacy noise is not supported with Prio3CountVec"
                                    .into(),
                            ));
                        }
                        Vdaf::CountVec {
                            length: length.unwrap(),
                            chunk_length,
                        }
                    }
                    VdafName::SumVec => {
                        let dp_strategy =
                            match (differential_privacy_strategy, differential_privacy_epsilon) {
                                (None, None) => dp_strategy::Prio3SumVec::NoDifferentialPrivacy,
                                (None, Some(_)) => {
                                    return Err(Error::Other(
                                        "missing differential-privacy-strategy".into(),
                                    ));
                                }
                                (Some(_), None) => {
                                    return Err(Error::Other(
                                        "missing differential-privacy-epsilon".into(),
                                    ));
                                }
                                (Some(DpStrategy::PureDpDiscreteLaplace), Some(epsilon)) => {
                                    dp_strategy::Prio3SumVec::PureDpDiscreteLaplace(
                                        PureDpDiscreteLaplace {
                                            budget: PureDpBudget {
                                                epsilon: float_to_biguint_ratio(epsilon)
                                                    .ok_or_else(|| {
                                                        Error::Other("invalid epsilon".into())
                                                    })?,
                                            },
                                        },
                                    )
                                }
                            };
                        Vdaf::SumVec(SumVec::new(
                            bits.unwrap(),
                            length.unwrap(),
                            chunk_length,
                            dp_strategy,
                        ))
                    }
                };

                let time_precision_seconds = time_precision.as_secs();
                let batch_time_window_size_seconds =
                    batch_time_window_size.map(|window| window.as_secs());

                let task = NewTask {
                    name,
                    leader_aggregator_id,
                    helper_aggregator_id,
                    vdaf,
                    min_batch_size,
                    max_batch_size,
                    batch_time_window_size_seconds,
                    time_precision_seconds,
                    collector_credential_id,
                };

                output.display(client.create_task(account_id, task).await?)
            }

            TaskAction::Rename { task_id, name } => {
                output.display(client.rename_task(&task_id, &name).await?)
            }

            TaskAction::Delete { task_id, force } => {
                if force {
                    client.force_delete_task(&task_id).await?
                } else {
                    client.delete_task(&task_id).await?
                }
            }
            TaskAction::SetExpiration {
                task_id,
                expiration,
                now,
            } => {
                let expiration = if now {
                    Some(OffsetDateTime::now_utc())
                } else {
                    expiration.map(|e| OffsetDateTime::from(Into::<SystemTime>::into(e)))
                };
                output.display(
                    client
                        .set_task_expiration(&task_id, expiration.as_ref())
                        .await?,
                )
            }
        }

        Ok(())
    }
}

fn float_to_biguint_ratio(value: f64) -> Option<Ratio<BigUint>> {
    let signed_ratio = Ratio::from_float(value)?;
    let unsigned_ratio = Ratio::new(
        signed_ratio.numer().clone().try_into().ok()?,
        signed_ratio.denom().clone().try_into().ok()?,
    );
    Some(unsigned_ratio)
}

1	use crate::{CliResult, DetermineAccountId, Error, Output};
2	use clap::Subcommand;
3	use divviup_client::{
4	BigUint, DivviupClient, Histogram, NewTask, Ratio, SumVec, Uuid, Vdaf,
5	dp_strategy::{self, PureDpBudget, PureDpDiscreteLaplace},
6	};
7	use humantime::{Duration, Timestamp};
8	use std::time::SystemTime;
9	use time::OffsetDateTime;
10
11	#[derive(clap::ValueEnum, Clone, Debug)]
12	pub enum VdafName {
13	Count,
14	Histogram,
15	Sum,
16	CountVec,
17	SumVec,
18	}
19
20	#[derive(clap::ValueEnum, Clone, Debug)]
21	pub enum DpStrategy {
22	PureDpDiscreteLaplace,
23	}
24
25	#[derive(Subcommand, Debug)]
26	pub enum TaskAction {
27	/// list all tasks for the target account
28	List,
29
30	/// retrieve details of a single task. this also refreshes cached data, such as metrics.
31	Get { task_id: String },
32
33	/// create a new task for the target account
34	Create {
35	#[arg(long)]
36	name: String,
37	#[arg(long)]
38	leader_aggregator_id: Uuid,
39	#[arg(long)]
40	helper_aggregator_id: Uuid,
41	#[arg(long)]
42	vdaf: VdafName,
43	#[arg(long)]
44	min_batch_size: u64,
45	#[arg(long)]
46	max_batch_size: Option<u64>,
47	#[arg(long, requires = "max_batch_size")]
48	batch_time_window_size: Option<Duration>,
49	#[arg(long)]
50	time_precision: Duration,
51	#[arg(long)]
52	collector_credential_id: Uuid,
53	#[arg(long, value_delimiter = ',')]
54	categorical_buckets: Option<Vec<String>>,
55	#[arg(long, value_delimiter = ',')]
56	continuous_buckets: Option<Vec<u64>>,
57	#[arg(long, required_if_eq_any([("vdaf", "count_vec"), ("vdaf", "sum_vec")]))]
58	length: Option<u64>,
59	#[arg(long, required_if_eq_any([("vdaf", "sum"), ("vdaf", "sum_vec")]))]
60	bits: Option<u8>,
61	#[arg(long)]
62	chunk_length: Option<u64>,
63	#[arg(long, requires = "differential_privacy_epsilon")]
64	differential_privacy_strategy: Option<DpStrategy>,
65	#[arg(long, requires = "differential_privacy_strategy")]
66	differential_privacy_epsilon: Option<f64>,
67	},
68
69	/// rename a task
70	Rename { task_id: String, name: String },
71
72	/// delete a task
73	Delete {
74	task_id: String,
75	/// delete the task even if the aggregators are unreachable
76	#[arg(long, action)]
77	force: bool,
78	},
79
80	/// set the expiration date of a task
81	SetExpiration {
82	task_id: String,
83	/// the date and time to set to.
84	///
85	/// if omitted, unset the expiration. the format is RFC 3339.
86	expiration: Option<Timestamp>,
87	/// set the expiration to the current time, effectively disabling the task.
88	#[arg(long, action, conflicts_with = "expiration")]
89	now: bool,
90	},
91	}
92
93	impl TaskAction {
94	pub(crate) async fn run(	×
95	self,	×
96	account_id: DetermineAccountId,	×
97	client: DivviupClient,	×
98	output: Output,	×
99	) -> CliResult {	×
100	let account_id = account_id.await?;	×
101
102	match self {	×
103	TaskAction::List => output.display(client.tasks(account_id).await?),	×
104	TaskAction::Get { task_id } => output.display(client.task(&task_id).await?),	×
105	TaskAction::Create {
106	name,	×
107	leader_aggregator_id,	×
108	helper_aggregator_id,	×
109	vdaf,	×
110	min_batch_size,	×
111	max_batch_size,	×
112	batch_time_window_size,	×
113	time_precision,	×
114	collector_credential_id,	×
115	categorical_buckets,	×
116	continuous_buckets,	×
117	length,	×
118	bits,	×
119	chunk_length,	×
120	differential_privacy_strategy,	×
121	differential_privacy_epsilon,	×
122	} => {
123	let vdaf = match vdaf {	×
124	VdafName::Count => {
125	if differential_privacy_strategy.is_some()	×
126	\|\| differential_privacy_epsilon.is_some()	×
127	{
128	return Err(Error::Other(	×
129	"differential privacy noise is not yet supported with Prio3Count"	×
130	.into(),	×
131	));	×
132	}	×
133	Vdaf::Count	×
134	}
135	VdafName::Histogram => {
136	let dp_strategy =	×
137	match (differential_privacy_strategy, differential_privacy_epsilon) {	×
138	(None, None) => dp_strategy::Prio3Histogram::NoDifferentialPrivacy,	×
139	(None, Some(_)) => {
140	return Err(Error::Other(	×
141	"missing differential-privacy-strategy".into(),	×
NEW 142	));	×
143	}
144	(Some(_), None) => {
145	return Err(Error::Other(	×
146	"missing differential-privacy-epsilon".into(),	×
NEW 147	));	×
148	}
149	(Some(DpStrategy::PureDpDiscreteLaplace), Some(epsilon)) => {	×
150	dp_strategy::Prio3Histogram::PureDpDiscreteLaplace(
151	PureDpDiscreteLaplace {
152	budget: PureDpBudget {
153	epsilon: float_to_biguint_ratio(epsilon)	×
154	.ok_or_else(\|\| {	×
155	Error::Other("invalid epsilon".into())	×
156	})?,	×
157	},
158	},
159	)
160	}
161	};
162	match (length, categorical_buckets, continuous_buckets) {	×
163	(Some(length), None, None) => Vdaf::Histogram(Histogram::Length {	×
164	length,	×
165	chunk_length,	×
166	dp_strategy,	×
167	}),	×
168	(None, Some(buckets), None) => {	×
169	Vdaf::Histogram(Histogram::Categorical {	×
170	buckets,	×
171	chunk_length,	×
172	dp_strategy,	×
173	})	×
174	}
175	(None, None, Some(buckets)) => Vdaf::Histogram(Histogram::Continuous {	×
176	buckets,	×
177	chunk_length,	×
178	dp_strategy,	×
179	}),	×
180	(None, None, None) => {
181	return Err(Error::Other("continuous-buckets, categorical-buckets, or length are required for histogram vdaf".into()));	×
182	}
183	_ => {
184	return Err(Error::Other("continuous-buckets, categorical-buckets, and length are mutually exclusive".into()));	×
185	}
186	}
187	}
188	VdafName::Sum => {
189	if differential_privacy_strategy.is_some()	×
190	\|\| differential_privacy_epsilon.is_some()	×
191	{
192	return Err(Error::Other(	×
193	"differential privacy noise is not yet supported with Prio3Sum"	×
194	.into(),	×
195	));	×
196	}	×
197	Vdaf::Sum {	×
198	bits: bits.unwrap(),	×
199	}	×
200	}
201	VdafName::CountVec => {
202	if differential_privacy_strategy.is_some()	×
203	\|\| differential_privacy_epsilon.is_some()	×
204	{
205	return Err(Error::Other(	×
206	"differential privacy noise is not supported with Prio3CountVec"	×
207	.into(),	×
208	));	×
209	}	×
210	Vdaf::CountVec {	×
211	length: length.unwrap(),	×
212	chunk_length,	×
213	}	×
214	}
215	VdafName::SumVec => {
216	let dp_strategy =	×
217	match (differential_privacy_strategy, differential_privacy_epsilon) {	×
218	(None, None) => dp_strategy::Prio3SumVec::NoDifferentialPrivacy,	×
219	(None, Some(_)) => {
220	return Err(Error::Other(	×
221	"missing differential-privacy-strategy".into(),	×
NEW 222	));	×
223	}
224	(Some(_), None) => {
225	return Err(Error::Other(	×
226	"missing differential-privacy-epsilon".into(),	×
NEW 227	));	×
228	}
229	(Some(DpStrategy::PureDpDiscreteLaplace), Some(epsilon)) => {	×
230	dp_strategy::Prio3SumVec::PureDpDiscreteLaplace(
231	PureDpDiscreteLaplace {
232	budget: PureDpBudget {
233	epsilon: float_to_biguint_ratio(epsilon)	×
234	.ok_or_else(\|\| {	×
235	Error::Other("invalid epsilon".into())	×
236	})?,	×
237	},
238	},
239	)
240	}
241	};
242	Vdaf::SumVec(SumVec::new(	×
243	bits.unwrap(),	×
244	length.unwrap(),	×
245	chunk_length,	×
246	dp_strategy,	×
247	))	×
248	}
249	};
250
251	let time_precision_seconds = time_precision.as_secs();	×
252	let batch_time_window_size_seconds =	×
253	batch_time_window_size.map(\|window\| window.as_secs());	×
254
255	let task = NewTask {	×
256	name,	×
257	leader_aggregator_id,	×
258	helper_aggregator_id,	×
259	vdaf,	×
260	min_batch_size,	×
261	max_batch_size,	×
262	batch_time_window_size_seconds,	×
263	time_precision_seconds,	×
264	collector_credential_id,	×
265	};	×
266
267	output.display(client.create_task(account_id, task).await?)	×
268	}
269
270	TaskAction::Rename { task_id, name } => {	×
271	output.display(client.rename_task(&task_id, &name).await?)	×
272	}
273
274	TaskAction::Delete { task_id, force } => {	×
275	if force {	×
276	client.force_delete_task(&task_id).await?	×
277	} else {
278	client.delete_task(&task_id).await?	×
279	}
280	}
281	TaskAction::SetExpiration {
282	task_id,	×
283	expiration,	×
284	now,	×
285	} => {
286	let expiration = if now {	×
287	Some(OffsetDateTime::now_utc())	×
288	} else {
289	expiration.map(\|e\| OffsetDateTime::from(Into::<SystemTime>::into(e)))	×
290	};
291	output.display(	×
292	client	×
293	.set_task_expiration(&task_id, expiration.as_ref())	×
294	.await?,	×
295	)
296	}
297	}
298
299	Ok(())	×
300	}	×
301	}
302
303	fn float_to_biguint_ratio(value: f64) -> Option<Ratio<BigUint>> {	×
304	let signed_ratio = Ratio::from_float(value)?;	×
305	let unsigned_ratio = Ratio::new(	×
306	signed_ratio.numer().clone().try_into().ok()?,	×
307	signed_ratio.denom().clone().try_into().ok()?,	×
308	);
309	Some(unsigned_ratio)	×
310	}	×

divviup / divviup-api / 16810179837

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous