15912404487

Committed 26 Jun 2025 09:01PM UTC coverage: 57.019% (-0.07%) from 57.086%

Build # 15912404487

Build Type

Pull #1779

github

Committed by

web-flow

Commit Message

Merge 0d4094ab5 into 18a3faee7

Pull Request Pull Request #1779: Fix Clippy lint

Run Details

3871 of 6789 relevant lines covered (57.02%)

64.49 hits per line

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

95.65

/src/entity/task/provisionable_task.rs

use super::*;
use crate::{
    clients::aggregator_client::api_types::{AggregatorVdaf, AuthenticationToken, QueryType},
    entity::{Account, CollectorCredential, Protocol, Task},
    handler::Error,
    Crypter,
};
use sea_orm::IntoActiveModel;
use std::fmt::Debug;
use trillium_client::Client;

#[derive(Clone, Debug)]
pub struct ProvisionableTask {
    pub account: Account,
    pub id: String,
    pub vdaf_verify_key: String,
    pub name: String,
    pub leader_aggregator: Aggregator,
    pub helper_aggregator: Aggregator,
    pub vdaf: Vdaf,
    pub aggregator_vdaf: AggregatorVdaf,
    pub min_batch_size: u64,
    pub max_batch_size: Option<u64>,
    pub batch_time_window_size_seconds: Option<u64>,
    pub expiration: Option<OffsetDateTime>,
    pub time_precision_seconds: u64,
    pub collector_credential: CollectorCredential,
    pub aggregator_auth_token: Option<String>,
    pub protocol: Protocol,
}

impl ProvisionableTask {
    async fn provision_aggregator(
        &self,
        http_client: Client,
        aggregator: Aggregator,
        crypter: &Crypter,
    ) -> Result<TaskResponse, Error> {
        let response = aggregator
            .client(http_client, crypter)?
            .create_task(self)
            .await?;

        assert_same(&self.aggregator_vdaf, &response.vdaf, "vdaf")?;
        assert_same(
            self.min_batch_size,
            response.min_batch_size,
            "min_batch_size",
        )?;
        assert_same(&self.query_type(), &response.query_type, "query_type")?;
        assert_same(
            // precision is lost in the round trip so we truncate our own
            self.expiration
                .map(|t| t.replace_millisecond(0))
                .transpose()?,
            response.task_expiration()?,
            "task_expiration",
        )?;
        assert_same(
            self.time_precision_seconds,
            response.time_precision.as_seconds(),
            "time_precision",
        )?;

        assert_same(&*self.id, &*response.task_id.to_string(), "task_id")?;

        // there are likely some more validations needed
        Ok(response)
    }

    pub async fn provision(
        mut self,
        client: Client,
        crypter: &Crypter,
    ) -> Result<ActiveModel, Error> {
        let helper = self
            .provision_aggregator(client.clone(), self.helper_aggregator.clone(), crypter)
            .await?;

        self.aggregator_auth_token = helper.aggregator_auth_token.map(AuthenticationToken::token);

        let _leader = self
            .provision_aggregator(client, self.leader_aggregator.clone(), crypter)
            .await?;

        Ok(Task {
            id: self.id,
            account_id: self.account.id,
            name: self.name,
            vdaf: self.vdaf.into(),
            min_batch_size: self.min_batch_size.try_into()?,
            max_batch_size: self.max_batch_size.map(TryInto::try_into).transpose()?,
            batch_time_window_size_seconds: self
                .batch_time_window_size_seconds
                .map(TryInto::try_into)
                .transpose()?,
            created_at: OffsetDateTime::now_utc(),
            updated_at: OffsetDateTime::now_utc(),
            deleted_at: None,
            time_precision_seconds: self.time_precision_seconds.try_into()?,
            report_count: 0,
            aggregate_collection_count: 0,
            expiration: self.expiration,
            leader_aggregator_id: self.leader_aggregator.id,
            helper_aggregator_id: self.helper_aggregator.id,
            collector_credential_id: self.collector_credential.id,
            report_counter_interval_collected: 0,
            report_counter_decode_failure: 0,
            report_counter_decrypt_failure: 0,
            report_counter_expired: 0,
            report_counter_outdated_key: 0,
            report_counter_success: 0,
            report_counter_too_early: 0,
            report_counter_task_expired: 0,
        }
        .into_active_model())
    }

    pub fn query_type(&self) -> QueryType {
        if let Some(max_batch_size) = self.max_batch_size {
            QueryType::FixedSize {
                max_batch_size,
                batch_time_window_size: self.batch_time_window_size_seconds,
            }
        } else {
            QueryType::TimeInterval
        }
    }
}

1	use super::*;
2	use crate::{
3	clients::aggregator_client::api_types::{AggregatorVdaf, AuthenticationToken, QueryType},
4	entity::{Account, CollectorCredential, Protocol, Task},
5	handler::Error,
6	Crypter,
7	};
8	use sea_orm::IntoActiveModel;
9	use std::fmt::Debug;
10	use trillium_client::Client;
11
12	#[derive(Clone, Debug)]
13	pub struct ProvisionableTask {
14	pub account: Account,
15	pub id: String,
16	pub vdaf_verify_key: String,
17	pub name: String,
18	pub leader_aggregator: Aggregator,
19	pub helper_aggregator: Aggregator,
20	pub vdaf: Vdaf,
21	pub aggregator_vdaf: AggregatorVdaf,
22	pub min_batch_size: u64,
23	pub max_batch_size: Option<u64>,
24	pub batch_time_window_size_seconds: Option<u64>,
25	pub expiration: Option<OffsetDateTime>,
26	pub time_precision_seconds: u64,
27	pub collector_credential: CollectorCredential,
28	pub aggregator_auth_token: Option<String>,
29	pub protocol: Protocol,
30	}
31
32	impl ProvisionableTask {
33	async fn provision_aggregator(	14✔
34	&self,	14✔
35	http_client: Client,	14✔
36	aggregator: Aggregator,	14✔
37	crypter: &Crypter,	14✔
38	) -> Result<TaskResponse, Error> {	14✔
39	let response = aggregator	14✔
40	.client(http_client, crypter)?	14✔
41	.create_task(self)	14✔
42	.await?;	14✔
43
44	assert_same(&self.aggregator_vdaf, &response.vdaf, "vdaf")?;	14✔
45	assert_same(	14✔
46	self.min_batch_size,	14✔
47	response.min_batch_size,	14✔
48	"min_batch_size",
49	)?;	×
50	assert_same(&self.query_type(), &response.query_type, "query_type")?;	14✔
51	assert_same(	14✔
52	// precision is lost in the round trip so we truncate our own
53	self.expiration	14✔
54	.map(\|t\| t.replace_millisecond(0))	14✔
55	.transpose()?,	14✔
56	response.task_expiration()?,	14✔
57	"task_expiration",
58	)?;	×
59	assert_same(	14✔
60	self.time_precision_seconds,	14✔
61	response.time_precision.as_seconds(),	14✔
62	"time_precision",
63	)?;	×
64
65	assert_same(&self.id, &response.task_id.to_string(), "task_id")?;	14✔
66
67	// there are likely some more validations needed
68	Ok(response)	14✔
69	}	14✔
70
71	pub async fn provision(	7✔
72	mut self,	7✔
73	client: Client,	7✔
74	crypter: &Crypter,	7✔
75	) -> Result<ActiveModel, Error> {	7✔
76	let helper = self	7✔
77	.provision_aggregator(client.clone(), self.helper_aggregator.clone(), crypter)	7✔
78	.await?;	7✔
79
80	self.aggregator_auth_token = helper.aggregator_auth_token.map(AuthenticationToken::token);	7✔
81
82	let _leader = self	7✔
83	.provision_aggregator(client, self.leader_aggregator.clone(), crypter)	7✔
84	.await?;	7✔
85
86	Ok(Task {
87	id: self.id,	7✔
88	account_id: self.account.id,	7✔
89	name: self.name,	7✔
90	vdaf: self.vdaf.into(),	7✔
91	min_batch_size: self.min_batch_size.try_into()?,	7✔
92	max_batch_size: self.max_batch_size.map(TryInto::try_into).transpose()?,	7✔
93	batch_time_window_size_seconds: self	7✔
94	.batch_time_window_size_seconds	7✔
95	.map(TryInto::try_into)	7✔
96	.transpose()?,	7✔
97	created_at: OffsetDateTime::now_utc(),	7✔
98	updated_at: OffsetDateTime::now_utc(),	7✔
99	deleted_at: None,	7✔
100	time_precision_seconds: self.time_precision_seconds.try_into()?,	7✔
101	report_count: 0,
102	aggregate_collection_count: 0,
103	expiration: self.expiration,	7✔
104	leader_aggregator_id: self.leader_aggregator.id,	7✔
105	helper_aggregator_id: self.helper_aggregator.id,	7✔
106	collector_credential_id: self.collector_credential.id,	7✔
107	report_counter_interval_collected: 0,
108	report_counter_decode_failure: 0,
109	report_counter_decrypt_failure: 0,
110	report_counter_expired: 0,
111	report_counter_outdated_key: 0,
112	report_counter_success: 0,
113	report_counter_too_early: 0,
114	report_counter_task_expired: 0,
115	}
116	.into_active_model())	7✔
117	}	7✔
118
119	pub fn query_type(&self) -> QueryType {	28✔
120	if let Some(max_batch_size) = self.max_batch_size {	28✔
121	QueryType::FixedSize {	4✔
122	max_batch_size,	4✔
123	batch_time_window_size: self.batch_time_window_size_seconds,	4✔
124	}	4✔
125	} else {
126	QueryType::TimeInterval	24✔
127	}
128	}	28✔
129	}

divviup / divviup-api / 15912404487

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