Repository: cloudflare/daphne
Pull request #335 (2023-06-02 10:23:42)

Simplify trait lifetimes with GATs
The main traits of the protocol were overly complex in terms of lifetimes. Rust
recently added GATs which are perfect for expressing what we need here.



Patch:
diff --git a/daphne/src/auth.rs b/daphne/src/auth.rs
index 7b0f50d..bd43927 100644
--- a/daphne/src/auth.rs
+++ b/daphne/src/auth.rs
@@ -51,21 +51,23 @@ impl AsRef<BearerToken> for BearerToken {
 
 /// A source of bearer tokens used for authorizing DAP requests.
 #[async_trait(?Send)]
-pub trait BearerTokenProvider<'a> {
+pub trait BearerTokenProvider {
     /// A reference to a bearer token owned by the provider.
-    type WrappedBearerToken: AsRef<BearerToken>;
+    type WrappedBearerToken<'a>: AsRef<BearerToken>
+    where
+        Self: 'a;
 
     /// Fetch the Leader's bearer token for the given task, if the task is recognized.
-    async fn get_leader_bearer_token_for(
-        &'a self,
-        task_id: &'a TaskId,
-    ) -> Result<Option<Self::WrappedBearerToken>, DapError>;
+    async fn get_leader_bearer_token_for<'s>(
+        &'s self,
+        task_id: &'s TaskId,
+    ) -> Result<Option<Self::WrappedBearerToken<'s>>, DapError>;
 
     /// Fetch the Collector's bearer token for the given task, if the task is recognized.
-    async fn get_collector_bearer_token_for(
-        &'a self,
-        task_id: &'a TaskId,
-    ) -> Result<Option<Self::WrappedBearerToken>, DapError>;
+    async fn get_collector_bearer_token_for<'s>(
+        &'s self,
+        task_id: &'s TaskId,
+    ) -> Result<Option<Self::WrappedBearerToken<'s>>, DapError>;
 
     /// Returns true if the given bearer token matches the leader token configured for the "taskprov" extension.
     fn is_taskprov_leader_bearer_token(&self, token: &BearerToken) -> bool;
@@ -75,11 +77,11 @@ pub trait BearerTokenProvider<'a> {
 
     /// Return a bearer token that can be used to authorize a request with the given task ID and
     /// media type.
-    async fn authorize_with_bearer_token(
-        &'a self,
-        task_id: &'a TaskId,
+    async fn authorize_with_bearer_token<'s>(
+        &'s self,
+        task_id: &'s TaskId,
         media_type: &DapMediaType,
-    ) -> Result<Self::WrappedBearerToken, DapError> {
+    ) -> Result<Self::WrappedBearerToken<'s>, DapError> {
         if matches!(media_type.sender(), Some(DapSender::Leader)) {
             let token = self
                 .get_leader_bearer_token_for(task_id)
@@ -100,8 +102,8 @@ pub trait BearerTokenProvider<'a> {
     /// Return `None` if the request is authorized. Otherwise return `Some(reason)`, where `reason`
     /// is the reason for the failure.
     async fn bearer_token_authorized<T: AsRef<BearerToken>>(
-        &'a self,
-        req: &'a DapRequest<T>,
+        &self,
+        req: &DapRequest<T>,
     ) -> Result<Option<String>, DapError> {
         if req.task_id.is_none() {
             // Can't authorize request with missing task ID.
diff --git a/daphne/src/hpke.rs b/daphne/src/hpke.rs
index af8d176..207fa97 100644
--- a/daphne/src/hpke.rs
+++ b/daphne/src/hpke.rs
@@ -91,16 +91,18 @@ impl HpkeConfig {
 
 /// HPKE decrypter functionality.
 #[async_trait(?Send)]
-pub trait HpkeDecrypter<'a> {
+pub trait HpkeDecrypter {
     /// Return type of `get_hpke_config_for()`, wraps a reference to an HPKE config.
-    type WrappedHpkeConfig: AsRef<HpkeConfig>;
+    type WrappedHpkeConfig<'a>: AsRef<HpkeConfig>
+    where
+        Self: 'a;
 
     /// Look up the HPKE configuration to use for the given task ID (if specified).
-    async fn get_hpke_config_for(
-        &'a self,
+    async fn get_hpke_config_for<'s>(
+        &'s self,
         version: DapVersion,
         task_id: Option<&TaskId>,
-    ) -> Result<Self::WrappedHpkeConfig, DapError>;
+    ) -> Result<Self::WrappedHpkeConfig<'s>, DapError>;
 
     /// Returns `true` if a ciphertext with the HPKE config ID can be consumed in the current task.
     async fn can_hpke_decrypt(&self, task_id: &TaskId, config_id: u8) -> Result<bool, DapError>;
@@ -202,14 +204,14 @@ impl TryFrom<(HpkeConfig, HpkePrivateKey)> for HpkeReceiverConfig {
 }
 
 #[async_trait(?Send)]
-impl<'a> HpkeDecrypter<'a> for HpkeReceiverConfig {
-    type WrappedHpkeConfig = HpkeConfig;
+impl HpkeDecrypter for HpkeReceiverConfig {
+    type WrappedHpkeConfig<'a> = HpkeConfig;
 
-    async fn get_hpke_config_for(
-        &'a self,
+    async fn get_hpke_config_for<'s>(
+        &'s self,
         _version: DapVersion,
         _task_id: Option<&TaskId>,
-    ) -> Result<Self::WrappedHpkeConfig, DapError> {
+    ) -> Result<Self::WrappedHpkeConfig<'s>, DapError> {
         unreachable!("not implemented");
     }
 
diff --git a/daphne/src/lib.rs b/daphne/src/lib.rs
index f67929b..d83722e 100644
--- a/daphne/src/lib.rs
+++ b/daphne/src/lib.rs
@@ -420,11 +420,11 @@ impl DapTaskConfig {
     }
 
     /// Return the batch span of a set of reports with the given metadata.
-    pub fn batch_span_for_meta<'a>(
+    pub fn batch_span_for_meta<'sel, 'rep>(
         &self,
-        part_batch_sel: &'a PartialBatchSelector,
-        report_meta: impl Iterator<Item = &'a ReportMetadata>,
-    ) -> Result<HashMap<DapBatchBucket<'a>, Vec<&'a ReportMetadata>>, DapError> {
+        part_batch_sel: &'sel PartialBatchSelector,
+        report_meta: impl Iterator<Item = &'rep ReportMetadata>,
+    ) -> Result<HashMap<DapBatchBucket<'sel>, Vec<&'rep ReportMetadata>>, DapError> {
         if !self.query.is_valid_part_batch_sel(part_batch_sel) {
             return Err(DapError::fatal(
                 "partial batch selector not compatible with task",
diff --git a/daphne/src/roles.rs b/daphne/src/roles.rs
index 90042cf..638d072 100644
--- a/daphne/src/roles.rs
+++ b/daphne/src/roles.rs
@@ -40,12 +40,9 @@ pub trait DapAuthorizedSender<S> {
 
 /// DAP Aggregator functionality.
 #[async_trait(?Send)]
-pub trait DapAggregator<'srv, 'req, S>: HpkeDecrypter<'srv> + Sized
-where
-    'srv: 'req,
-{
+pub trait DapAggregator<S>: HpkeDecrypter + Sized {
     /// A refernce to a task configuration stored by the Aggregator.
-    type WrappedDapTaskConfig: AsRef<DapTaskConfig>;
+    type WrappedDapTaskConfig<'a>: AsRef<DapTaskConfig>;
 
     /// Decide whether the given DAP request is authorized.
     ///
@@ -79,16 +76,16 @@ where
     /// `get_task_config_for()` will return the configure task. Otherwise this call will return
     /// nothing.
     async fn taskprov_put(
-        &'srv self,
-        req: &'req DapRequest<S>,
+        &self,
+        req: &DapRequest<S>,
         task_config: DapTaskConfig,
     ) -> Result<(), DapError>;
 
     /// Look up the DAP task configuration for the given task ID.
-    async fn get_task_config_for(
-        &'srv self,
+    async fn get_task_config_for<'req>(
+        &self,
         task_id: Cow<'req, TaskId>,
-    ) -> Result<Option<Self::WrappedDapTaskConfig>, DapError>;
+    ) -> Result<Option<Self::WrappedDapTaskConfig<'req>>, DapError>;
 
     /// Get the current time (number of seconds since the beginning of UNIX time).
     fn get_current_time(&self) -> Time;
@@ -123,11 +120,11 @@ where
     /// Ensure a set of reorts can be aggregated. Return a transition failure for each report
     /// that must be rejected early, due to the repot being replayed, the bucket that contains the
     /// report being collected, etc.
-    async fn check_early_reject<'b>(
+    async fn check_early_reject(
         &self,
         task_id: &TaskId,
-        part_batch_sel: &'b PartialBatchSelector,
-        report_meta: impl Iterator<Item = &'b ReportMetadata>,
+        part_batch_sel: &PartialBatchSelector,
+        report_meta: impl Iterator<Item = &ReportMetadata>,
     ) -> Result<HashMap<ReportId, TransitionFailure>, DapError>;
 
     /// Mark a batch as collected.
@@ -138,10 +135,7 @@ where
     ) -> Result<(), DapError>;
 
     /// Handle HTTP GET to `/hpke_config?task_id=<task_id>`.
-    async fn http_get_hpke_config(
-        &'srv self,
-        req: &DapRequest<S>,
-    ) -> Result<DapResponse, DapAbort> {
+    async fn http_get_hpke_config(&self, req: &DapRequest<S>) -> Result<DapResponse, DapAbort> {
         // Check whether the DAP version indicated by the sender is supported.
         if req.version == DapVersion::Unknown {
             return Err(DapAbort::version_unknown());
@@ -252,10 +246,7 @@ macro_rules! leader_post {
 
 /// DAP Leader functionality.
 #[async_trait(?Send)]
-pub trait DapLeader<'srv, 'req, S>: DapAuthorizedSender<S> + DapAggregator<'srv, 'req, S>
-where
-    'srv: 'req,
-{
+pub trait DapLeader<S>: DapAuthorizedSender<S> + DapAggregator<S> {
     /// Data type used to guide selection of a set of reports for aggregation.
     type ReportSelector;
 
@@ -308,7 +299,7 @@ where
 
     /// Handle HTTP POST to `/upload`. The input is the encoded report sent in the body of the HTTP
     /// request.
-    async fn http_post_upload(&'srv self, req: &'req DapRequest<S>) -> Result<(), DapAbort> {
+    async fn http_post_upload(&self, req: &DapRequest<S>) -> Result<(), DapAbort> {
         let metrics = self.metrics().with_host(req.host());
         let task_id = req.task_id()?;
         debug!("upload for task {task_id}");
@@ -371,7 +362,7 @@ where
     /// Handle HTTP POST to `/collect`. The input is a [`CollectReq`](crate::messages::CollectReq).
     /// The return value is a URI that the Collector can poll later on to get the corresponding
     /// [`CollectResp`](crate::messages::CollectResp).
-    async fn http_post_collect(&'srv self, req: &'req DapRequest<S>) -> Result<Url, DapAbort> {
+    async fn http_post_collect(&self, req: &DapRequest<S>) -> Result<Url, DapAbort> {
         let now = self.get_current_time();
         let metrics = self.metrics().with_host(req.host());
         let task_id = req.task_id()?;
@@ -699,7 +690,7 @@ where
     /// create a bottleneck. Such deployments can improve throughput by running many aggregation
     /// jobs in parallel.
     async fn process(
-        &'srv self,
+        &self,
         selector: &Self::ReportSelector,
         host: &str,
     ) -> Result<DapLeaderProcessTelemetry, DapAbort> {
@@ -760,10 +751,7 @@ where
 
 /// DAP Helper functionality.
 #[async_trait(?Send)]
-pub trait DapHelper<'srv, 'req, S>: DapAggregator<'srv, 'req, S>
-where
-    'srv: 'req,
-{
+pub trait DapHelper<S>: DapAggregator<S> {
     /// Store the Helper's aggregation-flow state unless it already exists. Returns a boolean
     /// indicating if the operation succeeded.
     async fn put_helper_state_if_not_exists(
@@ -785,10 +773,7 @@ where
     /// AggregationJobContinueReq and the response is an AggregationJobResp.
     ///
     /// This is called during the Initialization and Continuation phases.
-    async fn http_post_aggregate(
-        &'srv self,
-        req: &'req DapRequest<S>,
-    ) -> Result<DapResponse, DapAbort> {
+    async fn http_post_aggregate(&self, req: &DapRequest<S>) -> Result<DapResponse, DapAbort> {
         let metrics = self.metrics().with_host(req.host());
         let task_id = req.task_id()?;
 
@@ -1047,8 +1032,8 @@ where
     ///
     /// This is called during the Collection phase.
     async fn http_post_aggregate_share(
-        &'srv self,
-        req: &'req DapRequest<S>,
+        &self,
+        req: &DapRequest<S>,
     ) -> Result<DapResponse, DapAbort> {
         let now = self.get_current_time();
         let metrics = self.metrics().with_host(req.host());
@@ -1177,17 +1162,14 @@ fn check_part_batch(
     Ok(())
 }
 
-async fn check_batch<'srv, 'req, S>(
-    agg: &impl DapAggregator<'srv, 'req, S>,
+async fn check_batch<S>(
+    agg: &impl DapAggregator<S>,
     task_config: &DapTaskConfig,
     task_id: &TaskId,
     batch_sel: &BatchSelector,
     agg_param: &[u8],
     now: Time,
-) -> Result<(), DapAbort>
-where
-    'srv: 'req,
-{
+) -> Result<(), DapAbort> {
     let global_config = agg.get_global_config();
     let batch_overlapping = agg.is_batch_overlapping(task_id, batch_sel);
 
@@ -1319,15 +1301,12 @@ fn check_request_content_type<S>(
     }
 }
 
-async fn resolve_taskprov<'srv, 'req, S>(
-    agg: &'srv impl DapAggregator<'srv, 'req, S>,
-    task_id: &'req TaskId,
-    req: &'req DapRequest<S>,
+async fn resolve_taskprov<S>(
+    agg: &impl DapAggregator<S>,
+    task_id: &TaskId,
+    req: &DapRequest<S>,
     report_metadata_advertisement: Option<&ReportMetadata>,
-) -> Result<(), DapError>
-where
-    'srv: 'req,
-{
+) -> Result<(), DapError> {
     if agg
         .get_task_config_for(Cow::Borrowed(task_id))
         .await?
diff --git a/daphne/src/vdaf/mod.rs b/daphne/src/vdaf/mod.rs
index ec57fb6..2c66491 100644
--- a/daphne/src/vdaf/mod.rs
+++ b/daphne/src/vdaf/mod.rs
@@ -340,7 +340,7 @@ impl VdafConfig {
     #[allow(clippy::too_many_arguments)]
     pub(crate) async fn consume_report_share(
         &self,
-        decrypter: &impl HpkeDecrypter<'_>,
+        decrypter: &impl HpkeDecrypter,
         is_leader: bool,
         task_id: &TaskId,
         task_config: &DapTaskConfig,
@@ -420,7 +420,7 @@ impl VdafConfig {
     #[allow(clippy::too_many_arguments)]
     pub async fn produce_agg_job_init_req(
         &self,
-        decrypter: &impl HpkeDecrypter<'_>,
+        decrypter: &impl HpkeDecrypter,
         task_id: &TaskId,
         task_config: &DapTaskConfig,
         agg_job_id: &MetaAggregationJobId<'_>,
@@ -519,7 +519,7 @@ impl VdafConfig {
     /// * `version` is the DapVersion to use.
     pub(crate) async fn handle_agg_job_init_req(
         &self,
-        decrypter: &impl HpkeDecrypter<'_>,
+        decrypter: &impl HpkeDecrypter,
         task_id: &TaskId,
         task_config: &DapTaskConfig,
         agg_job_init_req: &AggregationJobInitReq,
@@ -885,7 +885,7 @@ impl VdafConfig {
     // TODO spec: Allow the collector to have multiple HPKE public keys (the way Aggregators do).
     pub async fn consume_encrypted_agg_shares(
         &self,
-        decrypter: &impl HpkeDecrypter<'_>,
+        decrypter: &impl HpkeDecrypter,
         task_id: &TaskId,
         batch_sel: &BatchSelector,
         report_count: u64,
diff --git a/daphne_worker/src/config.rs b/daphne_worker/src/config.rs
index e75ce99..cfea532 100644
--- a/daphne_worker/src/config.rs
+++ b/daphne_worker/src/config.rs
@@ -652,7 +652,7 @@ impl<'srv> DaphneWorker<'srv> {
     pub(crate) async fn get_leader_bearer_token<'a>(
         &'a self,
         task_id: &'a TaskId,
-    ) -> Result<Option<BearerTokenKvPair>> {
+    ) -> Result<Option<BearerTokenKvPair<'a>>> {
         self.kv_get_cached(
             &self.isolate_state().leader_bearer_tokens,
             KV_KEY_PREFIX_BEARER_TOKEN_LEADER,
@@ -686,12 +686,9 @@ impl<'srv> DaphneWorker<'srv> {
 
     /// Retrieve from KV the configuration for the given task.
     pub(crate) async fn get_task_config<'req>(
-        &'srv self,
+        &self,
         task_id: Cow<'req, TaskId>,
-    ) -> Result<Option<DapTaskConfigKvPair<'req>>>
-    where
-        'srv: 'req,
-    {
+    ) -> Result<Option<DapTaskConfigKvPair<'req>>> {
         self.kv_get_cached(
             &self.isolate_state().tasks,
             KV_KEY_PREFIX_TASK_CONFIG,
diff --git a/daphne_worker/src/dap.rs b/daphne_worker/src/dap.rs
index e95b5ea..ddd31d0 100644
--- a/daphne_worker/src/dap.rs
+++ b/daphne_worker/src/dap.rs
@@ -84,14 +84,15 @@ pub(crate) fn dap_response_to_worker(resp: DapResponse) -> Result<Response> {
 }
 
 #[async_trait(?Send)]
-impl<'srv> HpkeDecrypter<'srv> for DaphneWorker<'srv> {
-    type WrappedHpkeConfig = HpkeReceiverConfigKvPair<'srv>;
+impl<'srv> HpkeDecrypter for DaphneWorker<'srv> {
+    type WrappedHpkeConfig<'a> = HpkeReceiverConfigKvPair<'a>
+        where Self: 'a;
 
-    async fn get_hpke_config_for(
-        &'srv self,
+    async fn get_hpke_config_for<'s>(
+        &'s self,
         version: DapVersion,
         _task_id: Option<&TaskId>,
-    ) -> std::result::Result<HpkeReceiverConfigKvPair<'srv>, DapError> {
+    ) -> std::result::Result<HpkeReceiverConfigKvPair<'s>, DapError> {
         let kv_store = self.kv().map_err(dap_err)?;
         let keys = kv_store
             .list()
@@ -204,20 +205,21 @@ impl<'srv> HpkeDecrypter<'srv> for DaphneWorker<'srv> {
 }
 
 #[async_trait(?Send)]
-impl<'srv> BearerTokenProvider<'srv> for DaphneWorker<'srv> {
-    type WrappedBearerToken = BearerTokenKvPair<'srv>;
-
-    async fn get_leader_bearer_token_for(
-        &'srv self,
-        task_id: &'srv TaskId,
-    ) -> std::result::Result<Option<BearerTokenKvPair>, DapError> {
+impl<'srv> BearerTokenProvider for DaphneWorker<'srv> {
+    type WrappedBearerToken<'a> = BearerTokenKvPair<'a>
+        where Self: 'a;
+
+    async fn get_leader_bearer_token_for<'s>(
+        &'s self,
+        task_id: &'s TaskId,
+    ) -> std::result::Result<Option<Self::WrappedBearerToken<'s>>, DapError> {
         self.get_leader_bearer_token(task_id).await.map_err(dap_err)
     }
 
-    async fn get_collector_bearer_token_for(
-        &'srv self,
-        task_id: &'srv TaskId,
-    ) -> std::result::Result<Option<BearerTokenKvPair>, DapError> {
+    async fn get_collector_bearer_token_for<'s>(
+        &'s self,
+        task_id: &'s TaskId,
+    ) -> std::result::Result<Option<Self::WrappedBearerToken<'s>>, DapError> {
         self.get_collector_bearer_token(task_id)
             .await
             .map_err(dap_err)
@@ -270,11 +272,8 @@ impl DapAuthorizedSender<DaphneWorkerAuth> for DaphneWorker<'_> {
 }
 
 #[async_trait(?Send)]
-impl<'srv, 'req> DapAggregator<'srv, 'req, DaphneWorkerAuth> for DaphneWorker<'srv>
-where
-    'srv: 'req,
-{
-    type WrappedDapTaskConfig = DapTaskConfigKvPair<'req>;
+impl<'srv> DapAggregator<DaphneWorkerAuth> for DaphneWorker<'srv> {
+    type WrappedDapTaskConfig<'a> = DapTaskConfigKvPair<'a>;
 
     async fn unauthorized_reason(
         &self,
@@ -376,8 +375,8 @@ where
     }
 
     async fn taskprov_put(
-        &'srv self,
-        req: &'req DapRequest<DaphneWorkerAuth>,
+        &self,
+        req: &DapRequest<DaphneWorkerAuth>,
         task_config: DapTaskConfig,
     ) -> std::result::Result<(), DapError> {
         let task_id = req.task_id().map_err(DapError::Abort)?;
@@ -429,10 +428,10 @@ where
         Ok(())
     }
 
-    async fn get_task_config_for(
-        &'srv self,
+    async fn get_task_config_for<'req>(
+        &self,
         task_id: Cow<'req, TaskId>,
-    ) -> std::result::Result<Option<DapTaskConfigKvPair<'req>>, DapError> {
+    ) -> std::result::Result<Option<Self::WrappedDapTaskConfig<'req>>, DapError> {
         self.get_task_config(task_id).await.map_err(dap_err)
     }
 
@@ -552,11 +551,11 @@ where
         Ok(agg_share)
     }
 
-    async fn check_early_reject<'b>(
+    async fn check_early_reject(
         &self,
         task_id: &TaskId,
-        part_batch_sel: &'b PartialBatchSelector,
-        report_meta: impl Iterator<Item = &'b ReportMetadata>,
+        part_batch_sel: &PartialBatchSelector,
+        report_meta: impl Iterator<Item = &ReportMetadata>,
     ) -> std::result::Result<HashMap<ReportId, TransitionFailure>, DapError> {
         let durable = self.durable().with_retry();
         let task_config = self.try_get_task_config(task_id).await?;
@@ -701,10 +700,7 @@ where
 }
 
 #[async_trait(?Send)]
-impl<'srv, 'req> DapLeader<'srv, 'req, DaphneWorkerAuth> for DaphneWorker<'srv>
-where
-    'srv: 'req,
-{
+impl<'srv> DapLeader<DaphneWorkerAuth> for DaphneWorker<'srv> {
     type ReportSelector = DaphneWorkerReportSelector;
 
     async fn put_report(
@@ -988,10 +984,7 @@ where
 }
 
 #[async_trait(?Send)]
-impl<'srv, 'req> DapHelper<'srv, 'req, DaphneWorkerAuth> for DaphneWorker<'srv>
-where
-    'srv: 'req,
-{
+impl<'srv> DapHelper<DaphneWorkerAuth> for DaphneWorker<'srv> {
     async fn put_helper_state_if_not_exists(
         &self,
         task_id: &TaskId,


Test patch:
diff --git a/daphne/src/testing.rs b/daphne/src/testing.rs
index 545861d..f8abb35 100644
--- a/daphne/src/testing.rs
+++ b/daphne/src/testing.rs
@@ -224,20 +224,20 @@ impl MockAggregator {
 }
 
 #[async_trait(?Send)]
-impl<'a> BearerTokenProvider<'a> for MockAggregator {
-    type WrappedBearerToken = &'a BearerToken;
+impl BearerTokenProvider for MockAggregator {
+    type WrappedBearerToken<'a> = &'a BearerToken;
 
-    async fn get_leader_bearer_token_for(
-        &'a self,
-        _task_id: &'a TaskId,
-    ) -> Result<Option<&'a BearerToken>, DapError> {
+    async fn get_leader_bearer_token_for<'s>(
+        &'s self,
+        _task_id: &'s TaskId,
+    ) -> Result<Option<Self::WrappedBearerToken<'s>>, DapError> {
         Ok(Some(&self.leader_token))
     }
 
-    async fn get_collector_bearer_token_for(
-        &'a self,
-        _task_id: &'a TaskId,
-    ) -> Result<Option<&'a BearerToken>, DapError> {
+    async fn get_collector_bearer_token_for<'s>(
+        &'s self,
+        _task_id: &'s TaskId,
+    ) -> Result<Option<Self::WrappedBearerToken<'s>>, DapError> {
         if let Some(ref collector_token) = self.collector_token {
             Ok(Some(collector_token))
         } else {
@@ -263,14 +263,14 @@ impl<'a> BearerTokenProvider<'a> for MockAggregator {
 }
 
 #[async_trait(?Send)]
-impl<'a> HpkeDecrypter<'a> for MockAggregator {
-    type WrappedHpkeConfig = &'a HpkeConfig;
+impl HpkeDecrypter for MockAggregator {
+    type WrappedHpkeConfig<'a> = &'a HpkeConfig;
 
-    async fn get_hpke_config_for(
-        &'a self,
+    async fn get_hpke_config_for<'s>(
+        &'s self,
         _version: DapVersion,
         task_id: Option<&TaskId>,
-    ) -> Result<&'a HpkeConfig, DapError> {
+    ) -> Result<Self::WrappedHpkeConfig<'s>, DapError> {
         if self.hpke_receiver_config_list.is_empty() {
             return Err(DapError::fatal("emtpy HPKE receiver config list"));
         }
@@ -324,14 +324,11 @@ impl DapAuthorizedSender<BearerToken> for MockAggregator {
 }
 
 #[async_trait(?Send)]
-impl<'srv, 'req> DapAggregator<'srv, 'req, BearerToken> for MockAggregator
-where
-    'srv: 'req,
-{
+impl DapAggregator<BearerToken> for MockAggregator {
     // The lifetimes on the traits ensure that we can return a reference to a task config stored by
     // the DapAggregator. (See DaphneWorkerConfig for an example.) For simplicity, MockAggregator
     // clones the task config as needed.
-    type WrappedDapTaskConfig = DapTaskConfig;
+    type WrappedDapTaskConfig<'a> = DapTaskConfig;
 
     async fn unauthorized_reason(
         &self,
@@ -361,8 +358,8 @@ where
     }
 
     async fn taskprov_put(
-        &'srv self,
-        req: &'req DapRequest<BearerToken>,
+        &self,
+        req: &DapRequest<BearerToken>,
         task_config: DapTaskConfig,
     ) -> Result<(), DapError> {
         let task_id = req.task_id().map_err(DapError::Abort)?;
@@ -371,10 +368,10 @@ where
         Ok(())
     }
 
-    async fn get_task_config_for(
-        &'srv self,
-        task_id: Cow<'req, TaskId>,
-    ) -> Result<Option<Self::WrappedDapTaskConfig>, DapError> {
+    async fn get_task_config_for<'s>(
+        &self,
+        task_id: Cow<'s, TaskId>,
+    ) -> Result<Option<Self::WrappedDapTaskConfig<'s>>, DapError> {
         let tasks = self.tasks.lock().expect("tasks: lock failed");
         Ok(tasks.get(task_id.as_ref()).cloned())
     }
@@ -479,11 +476,11 @@ where
         Ok(agg_share)
     }
 
-    async fn check_early_reject<'b>(
+    async fn check_early_reject(
         &self,
         task_id: &TaskId,
-        part_batch_sel: &'b PartialBatchSelector,
-        report_meta: impl Iterator<Item = &'b ReportMetadata>,
+        part_batch_sel: &PartialBatchSelector,
+        report_meta: impl Iterator<Item = &ReportMetadata>,
     ) -> Result<HashMap<ReportId, TransitionFailure>, DapError> {
         let task_config = self
             .get_task_config_for(Cow::Borrowed(task_id))
@@ -550,10 +547,7 @@ where
 }
 
 #[async_trait(?Send)]
-impl<'srv, 'req> DapHelper<'srv, 'req, BearerToken> for MockAggregator
-where
-    'srv: 'req,
-{
+impl DapHelper<BearerToken> for MockAggregator {
     async fn put_helper_state_if_not_exists(
         &self,
         task_id: &TaskId,
@@ -613,10 +607,7 @@ where
 }
 
 #[async_trait(?Send)]
-impl<'srv, 'req> DapLeader<'srv, 'req, BearerToken> for MockAggregator
-where
-    'srv: 'req,
-{
+impl DapLeader<BearerToken> for MockAggregator {
     type ReportSelector = MockAggregatorReportSelector;
 
     async fn put_report(&self, report: &Report, task_id: &TaskId) -> Result<(), DapError> {