Remove DapReportInitializer

crates/dapf/Cargo.toml

@@ -13,7 +13,6 @@ repository.workspace = true
 
 [dependencies]
 anyhow.workspace = true
-async-trait.workspace = true
 chrono.workspace = true
 clap = { workspace = true, features = ["env"] }
 daphne = { path = "../daphne", features = ["report-generator", "prometheus"] }

crates/dapf/src/acceptance/mod.rs

@@ -20,15 +20,13 @@ pub mod load_testing;
 
 use crate::{deduce_dap_version_from_url, functions, test_durations::TestDurations, HttpClient};
 use anyhow::{anyhow, bail, Context, Result};
-use async_trait::async_trait;
 use daphne::{
     hpke::{HpkeConfig, HpkeKemId, HpkeReceiverConfig},
     messages::{
         self, taskprov::TaskprovAdvertisement, AggregateShareReq, AggregationJobId, Base64Encode,
         BatchId, BatchSelector, PartialBatchSelector, TaskId,
     },
     metrics::DaphneMetrics,
-    roles::DapReportInitializer,
     testing::report_generator::ReportGenerator,
     vdaf::VdafConfig,
     DapAggregateShare, DapAggregateSpan, DapAggregationParam, DapMeasurement, DapQueryConfig,
@@ -42,7 +40,6 @@ use std::{
     convert::TryFrom,
     env,
     num::NonZeroU32,
-    ops::Range,
     path::PathBuf,
     sync::atomic::{AtomicUsize, Ordering},
     time::{Duration, Instant, SystemTime},
@@ -489,7 +486,7 @@ impl Test {
         let (agg_job_state, agg_job_init_req) = task_config
             .test_produce_agg_job_req(
                 fake_leader_hpke_receiver_config,
-                self,
+                messages::Time::MIN..messages::Time::MAX,
                 task_id,
                 part_batch_sel,
                 &DapAggregationParam::Empty,
@@ -697,14 +694,6 @@ impl Test {
     }
 }
 
-#[async_trait]
-impl DapReportInitializer for Test {
-    fn valid_report_time_range(&self) -> Range<messages::Time> {
-        // Accept reports with any timestmap.
-        0..u64::MAX
-    }
-}
-
 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub struct Now(u64);
 pub fn now() -> Now {

crates/daphne-server/src/roles/aggregator.rs

@@ -13,7 +13,7 @@ use daphne::{
     metrics::DaphneMetrics,
     roles::{
         aggregator::{MergeAggShareError, TaskprovConfig},
-        DapAggregator, DapReportInitializer,
+        DapAggregator,
     },
     taskprov, DapAggregateShare, DapAggregateSpan, DapError, DapGlobalConfig, DapTaskConfig,
     DapVersion,
@@ -344,10 +344,7 @@ impl DapAggregator for crate::App {
     fn audit_log(&self) -> &dyn AuditLog {
         &*self.audit_log
     }
-}
 
-#[async_trait]
-impl DapReportInitializer for crate::App {
     fn valid_report_time_range(&self) -> Range<messages::Time> {
         let now = SystemTime::now()
             .duration_since(SystemTime::UNIX_EPOCH)

crates/daphne/src/protocol/aggregator.rs

@@ -14,7 +14,6 @@ use crate::{
         Transition, TransitionFailure, TransitionVar,
     },
     metrics::{DaphneMetrics, ReportStatus},
-    roles::DapReportInitializer,
     vdaf::{
         prio2::{prio2_prep_finish, prio2_prep_finish_from_shares, prio2_prep_init},
         prio3::{prio3_prep_finish, prio3_prep_finish_from_shares, prio3_prep_init},
@@ -294,7 +293,7 @@ impl DapTaskConfig {
     pub fn produce_agg_job_req<S>(
         &self,
         decrypter: impl HpkeDecrypter,
-        initializer: &impl DapReportInitializer,
+        valid_report_time_range: Range<messages::Time>,
         task_id: &TaskId,
         part_batch_sel: &PartialBatchSelector,
         agg_param: &DapAggregationParam,
@@ -306,7 +305,7 @@ impl DapTaskConfig {
     {
         self.produce_agg_job_req_impl(
             decrypter,
-            initializer,
+            valid_report_time_range,
             task_id,
             part_batch_sel,
             agg_param,
@@ -320,7 +319,7 @@ impl DapTaskConfig {
     fn produce_agg_job_req_impl<S>(
         &self,
         decrypter: impl HpkeDecrypter,
-        initializer: &impl DapReportInitializer,
+        valid_report_time_range: Range<messages::Time>,
         task_id: &TaskId,
         part_batch_sel: &PartialBatchSelector,
         agg_param: &DapAggregationParam,
@@ -354,7 +353,7 @@ impl DapTaskConfig {
 
             let initialized_report = InitializedReport::new(
                 &decrypter,
-                initializer.valid_report_time_range(),
+                valid_report_time_range.clone(),
                 true,
                 task_id,
                 self,
@@ -430,7 +429,7 @@ impl DapTaskConfig {
     pub fn test_produce_agg_job_req<S>(
         &self,
         decrypter: impl HpkeDecrypter,
-        initializer: &impl DapReportInitializer,
+        valid_report_time_range: Range<messages::Time>,
         task_id: &TaskId,
         part_batch_sel: &PartialBatchSelector,
         agg_param: &DapAggregationParam,
@@ -443,7 +442,7 @@ impl DapTaskConfig {
     {
         self.produce_agg_job_req_impl(
             decrypter,
-            initializer,
+            valid_report_time_range,
             task_id,
             part_batch_sel,
             agg_param,
@@ -456,17 +455,16 @@ impl DapTaskConfig {
     /// Helper: Consume the `AggregationJobInitReq` sent by the Leader and return the initialized
     /// reports.
     #[tracing::instrument(skip_all)]
-    pub fn consume_agg_job_req<H, I>(
+    pub fn consume_agg_job_req<H>(
         &self,
         decrypter: &H,
-        initializer: &I,
+        valid_report_time_range: Range<messages::Time>,
         task_id: &TaskId,
         agg_job_init_req: AggregationJobInitReq,
         replay_protection: ReplayProtection,
     ) -> Result<Vec<InitializedReport>, DapError>
     where
         H: HpkeDecrypter + Sync,
-        I: DapReportInitializer + Sync,
     {
         let agg_param =
             DapAggregationParam::get_decoded_with_param(&self.vdaf, &agg_job_init_req.agg_param)
@@ -490,7 +488,7 @@ impl DapTaskConfig {
             .map(|prep_init| {
                 InitializedReport::new(
                     decrypter,
-                    initializer.valid_report_time_range(),
+                    valid_report_time_range.clone(),
                     false,
                     task_id,
                     self,

crates/daphne/src/protocol/mod.rs

@@ -24,7 +24,6 @@ mod test {
             TransitionVar,
         },
         protocol::aggregator::InitializedReport,
-        roles::DapReportInitializer,
         test_versions,
         testing::AggregationJobTest,
         vdaf::{Prio3Config, VdafConfig},

crates/daphne/src/roles/aggregator.rs

@@ -17,14 +17,6 @@ use crate::{
     DapTaskConfig, DapVersion,
 };
 
-/// Report initializer. Used by a DAP Aggregator [`DapAggregator`] when initializing an aggregation
-/// job.
-#[async_trait]
-pub trait DapReportInitializer {
-    /// Return the time range in which a report must appear in order to be considered valid.
-    fn valid_report_time_range(&self) -> Range<Time>;
-}
-
 #[derive(Debug)]
 pub enum MergeAggShareError {
     AlreadyCollected,
@@ -42,7 +34,7 @@ pub struct TaskprovConfig<'s> {
 
 /// DAP Aggregator functionality.
 #[async_trait]
-pub trait DapAggregator: HpkeProvider + DapReportInitializer + Sized {
+pub trait DapAggregator: HpkeProvider + Sized {
     /// Look up the DAP global configuration.
     async fn get_global_config(&self) -> Result<DapGlobalConfig, DapError>;
 
@@ -137,6 +129,9 @@ pub trait DapAggregator: HpkeProvider + DapReportInitializer + Sized {
 
     /// Access the audit log.
     fn audit_log(&self) -> &dyn AuditLog;
+
+    /// Return the time range in which a report must appear in order to be considered valid.
+    fn valid_report_time_range(&self) -> Range<Time>;
 }
 
 /// Handle request for the Aggregator's HPKE configuration.

crates/daphne/src/roles/helper.rs

@@ -48,7 +48,7 @@ pub async fn handle_agg_job_init_req<A: DapHelper + Sync>(
     let version = req.version;
     let initialized_reports = task_config.consume_agg_job_req(
         &aggregator.get_receiver_configs(task_config.version).await?,
-        aggregator,
+        aggregator.valid_report_time_range(),
         &task_id,
         req.payload,
         replay_protection,

crates/daphne/src/roles/leader/mod.rs

@@ -301,7 +301,7 @@ async fn run_agg_job<A: DapLeader>(
     let agg_job_id = AggregationJobId(thread_rng().gen());
     let (agg_job_state, agg_job_init_req) = task_config.produce_agg_job_req(
         aggregator.get_receiver_configs(task_config.version).await?,
-        aggregator,
+        aggregator.valid_report_time_range(),
         task_id,
         part_batch_sel,
         agg_param,

crates/daphne/src/roles/mod.rs

@@ -13,7 +13,7 @@ use crate::{
     DapAbort, DapError, DapGlobalConfig, DapQueryConfig, DapRequestMeta, DapTaskConfig,
 };
 
-pub use aggregator::{DapAggregator, DapReportInitializer};
+pub use aggregator::DapAggregator;
 pub use helper::DapHelper;
 pub use leader::DapLeader;
 
@@ -459,7 +459,7 @@ mod test {
             let (leader_state, agg_job_init_req) = task_config
                 .produce_agg_job_req(
                     &*self.leader.hpke_receiver_config_list,
-                    &*self.leader,
+                    self.leader.valid_report_time_range(),
                     task_id,
                     &part_batch_sel,
                     &agg_param,

crates/daphne/src/testing/mod.rs

@@ -21,7 +21,7 @@ use crate::{
         aggregator::{MergeAggShareError, TaskprovConfig},
         helper,
         leader::{in_memory_leader::InMemoryLeaderState, WorkItem},
-        DapAggregator, DapHelper, DapLeader, DapReportInitializer,
+        DapAggregator, DapHelper, DapLeader,
     },
     taskprov, DapAbort, DapAggregateResult, DapAggregateShare, DapAggregateSpan,
     DapAggregationJobState, DapAggregationParam, DapBatchBucket, DapCollectionJob, DapError,
@@ -68,14 +68,11 @@ pub struct AggregationJobTest {
     pub(crate) leader_metrics: DaphnePromMetrics,
 }
 
-#[async_trait]
-impl DapReportInitializer for AggregationJobTest {
-    fn valid_report_time_range(&self) -> Range<Time> {
+impl AggregationJobTest {
+    pub fn valid_report_time_range(&self) -> Range<Time> {
         self.valid_report_range.clone()
     }
-}
 
-impl AggregationJobTest {
     /// Create an aggregation job test with the given VDAF config, HPKE KEM algorithm, DAP protocol
     /// version. The KEM algorithm is used to generate an HPKE config for each party.
     pub fn new(vdaf: &VdafConfig, kem_id: HpkeKemId, version: DapVersion) -> Self {
@@ -193,7 +190,7 @@ impl AggregationJobTest {
         self.task_config
             .test_produce_agg_job_req(
                 &self.leader_hpke_receiver_config,
-                self,
+                self.valid_report_time_range(),
                 &self.task_id,
                 &PartialBatchSelector::TimeInterval,
                 agg_param,
@@ -219,7 +216,7 @@ impl AggregationJobTest {
                     .task_config
                     .consume_agg_job_req(
                         &self.helper_hpke_receiver_config,
-                        self,
+                        self.valid_report_time_range(),
                         &self.task_id,
                         agg_job_init_req,
                         self.replay_protection,
@@ -654,14 +651,6 @@ impl HpkeProvider for InMemoryAggregator {
     }
 }
 
-#[async_trait]
-impl DapReportInitializer for InMemoryAggregator {
-    fn valid_report_time_range(&self) -> Range<messages::Time> {
-        // Accept reports with any timestmap.
-        0..u64::MAX
-    }
-}
-
 #[async_trait]
 impl DapAggregator for InMemoryAggregator {
     async fn get_global_config(&self) -> Result<DapGlobalConfig, DapError> {
@@ -866,6 +855,11 @@ impl DapAggregator for InMemoryAggregator {
     fn audit_log(&self) -> &dyn AuditLog {
         &self.audit_log
     }
+
+    fn valid_report_time_range(&self) -> Range<messages::Time> {
+        // Accept reports with any timestmap.
+        messages::Time::MIN..messages::Time::MAX
+    }
 }
 
 #[async_trait]