Repository: cloudflare/daphne
Pull request #447 (2023-11-30 02:20:27)

taskprov: Enable Prio3SumVecField64MultiproofHmacSha256Aes128 in draft09
Stacked on #446.


Patch:
diff --git a/daphne/src/lib.rs b/daphne/src/lib.rs
index 6f1828f..c7cbc62 100644
--- a/daphne/src/lib.rs
+++ b/daphne/src/lib.rs
@@ -478,7 +478,7 @@ impl DapTaskParameters {
             task_expiration: now + 86400 * 14, // expires in two weeks
             vdaf_config: messages::taskprov::VdafConfig {
                 dp_config: messages::taskprov::DpConfig::None,
-                var: messages::taskprov::VdafTypeVar::Prio2 { dimension: 10 },
+                var: (&self.vdaf).try_into()?,
             },
         };
 
diff --git a/daphne/src/messages/mod.rs b/daphne/src/messages/mod.rs
index 198d190..ee3aaf8 100644
--- a/daphne/src/messages/mod.rs
+++ b/daphne/src/messages/mod.rs
@@ -1288,31 +1288,6 @@ fn decode_u16_prefixed<O>(
     Ok(decoded)
 }
 
-fn encode_u16_item_for_version<E: ParameterizedEncode<DapVersion>>(
-    bytes: &mut Vec<u8>,
-    version: DapVersion,
-    item: &E,
-) {
-    match version {
-        DapVersion::DraftLatest => encode_u16_prefixed(version, bytes, |version, bytes| {
-            item.encode_with_param(&version, bytes);
-        }),
-        DapVersion::Draft02 => item.encode_with_param(&version, bytes),
-    }
-}
-
-fn decode_u16_item_for_version<D: ParameterizedDecode<DapVersion>>(
-    version: DapVersion,
-    bytes: &mut Cursor<&[u8]>,
-) -> Result<D, CodecError> {
-    match version {
-        DapVersion::DraftLatest => decode_u16_prefixed(version, bytes, |version, inner| {
-            D::decode_with_param(&version, inner)
-        }),
-        DapVersion::Draft02 => D::decode_with_param(&version, bytes),
-    }
-}
-
 #[cfg(test)]
 mod test {
     use super::*;
diff --git a/daphne/src/messages/taskprov.rs b/daphne/src/messages/taskprov.rs
index d942580..41a1791 100644
--- a/daphne/src/messages/taskprov.rs
+++ b/daphne/src/messages/taskprov.rs
@@ -4,8 +4,8 @@
 //! draft-wang-ppm-dap-taskprov: Messages for the taskrpov extension for DAP.
 
 use crate::messages::{
-    decode_u16_bytes, decode_u16_item_for_version, encode_u16_bytes, encode_u16_item_for_version,
-    Duration, Time, QUERY_TYPE_FIXED_SIZE, QUERY_TYPE_TIME_INTERVAL,
+    decode_u16_bytes, encode_u16_bytes, Duration, Time, QUERY_TYPE_FIXED_SIZE,
+    QUERY_TYPE_TIME_INTERVAL,
 };
 use crate::DapVersion;
 use prio::codec::{
@@ -15,8 +15,11 @@ use prio::codec::{
 use serde::{Deserialize, Serialize};
 use std::io::Cursor;
 
+use super::{decode_u16_prefixed, encode_u16_prefixed};
+
 // VDAF type codes.
 const VDAF_TYPE_PRIO2: u32 = 0xFFFF_0000;
+pub(crate) const VDAF_TYPE_PRIO3_SUM_VEC_FIELD64_MULTIPROOF_HMAC_SHA256_AES128: u32 = 0xFFFF_1003;
 
 // Differential privacy mechanism types.
 const DP_MECHANISM_NONE: u8 = 0x01;
@@ -24,8 +27,19 @@ const DP_MECHANISM_NONE: u8 = 0x01;
 /// A VDAF type along with its type-specific data.
 #[derive(Clone, Deserialize, Serialize, Debug, PartialEq, Eq)]
 pub enum VdafTypeVar {
-    Prio2 { dimension: u32 },
-    NotImplemented { typ: u32, param: Vec<u8> },
+    Prio2 {
+        dimension: u32,
+    },
+    Prio3SumVecField64MultiproofHmacSha256Aes128 {
+        bits: u8,
+        length: u32,
+        chunk_length: u32,
+        num_proofs: u8,
+    },
+    NotImplemented {
+        typ: u32,
+        param: Vec<u8>,
+    },
 }
 
 impl ParameterizedEncode<DapVersion> for VdafTypeVar {
@@ -33,14 +47,29 @@ impl ParameterizedEncode<DapVersion> for VdafTypeVar {
         match self {
             Self::Prio2 { dimension } => {
                 VDAF_TYPE_PRIO2.encode(bytes);
-                encode_u16_item_for_version(bytes, *version, dimension);
+                taskprov_encode_u16_prefixed(*version, bytes, |_version, inner| {
+                    dimension.encode(inner);
+                });
+            }
+            Self::Prio3SumVecField64MultiproofHmacSha256Aes128 {
+                bits,
+                length,
+                chunk_length,
+                num_proofs,
+            } => {
+                VDAF_TYPE_PRIO3_SUM_VEC_FIELD64_MULTIPROOF_HMAC_SHA256_AES128.encode(bytes);
+                taskprov_encode_u16_prefixed(*version, bytes, |_version, inner| {
+                    bits.encode(inner);
+                    length.encode(inner);
+                    chunk_length.encode(inner);
+                    num_proofs.encode(inner);
+                });
             }
             Self::NotImplemented { typ, param } => {
                 typ.encode(bytes);
-                match version {
-                    DapVersion::DraftLatest => encode_u16_bytes(bytes, param),
-                    DapVersion::Draft02 => bytes.extend_from_slice(param),
-                }
+                taskprov_encode_u16_prefixed(*version, bytes, |_version, inner| {
+                    inner.extend_from_slice(param);
+                });
             }
         }
     }
@@ -53,9 +82,23 @@ impl ParameterizedDecode<DapVersion> for VdafTypeVar {
     ) -> Result<Self, CodecError> {
         let vdaf_type = u32::decode(bytes)?;
         match (version, vdaf_type) {
-            (.., VDAF_TYPE_PRIO2) => Ok(Self::Prio2 {
-                dimension: decode_u16_item_for_version(*version, bytes)?,
-            }),
+            (.., VDAF_TYPE_PRIO2) => {
+                taskprov_decode_u16_prefixed(*version, bytes, |_version, inner| {
+                    Ok(Self::Prio2 {
+                        dimension: u32::decode(inner)?,
+                    })
+                })
+            }
+            (.., VDAF_TYPE_PRIO3_SUM_VEC_FIELD64_MULTIPROOF_HMAC_SHA256_AES128) => {
+                taskprov_decode_u16_prefixed(*version, bytes, |_version, inner| {
+                    Ok(Self::Prio3SumVecField64MultiproofHmacSha256Aes128 {
+                        bits: u8::decode(inner)?,
+                        length: u32::decode(inner)?,
+                        chunk_length: u32::decode(inner)?,
+                        num_proofs: u8::decode(inner)?,
+                    })
+                })
+            }
             (DapVersion::DraftLatest, ..) => Ok(Self::NotImplemented {
                 typ: vdaf_type,
                 param: decode_u16_bytes(bytes)?,
@@ -79,15 +122,14 @@ impl ParameterizedEncode<DapVersion> for DpConfig {
         match self {
             Self::None => {
                 DP_MECHANISM_NONE.encode(bytes);
-                encode_u16_item_for_version(bytes, *version, &());
+                taskprov_encode_u16_prefixed(*version, bytes, |_, _| ());
             }
 
             Self::NotImplemented { typ, param } => {
                 typ.encode(bytes);
-                match version {
-                    DapVersion::DraftLatest => encode_u16_bytes(bytes, param),
-                    DapVersion::Draft02 => bytes.extend_from_slice(param),
-                }
+                taskprov_encode_u16_prefixed(*version, bytes, |_version, inner| {
+                    inner.extend_from_slice(param);
+                });
             }
         }
     }
@@ -101,7 +143,9 @@ impl ParameterizedDecode<DapVersion> for DpConfig {
         let dp_mechanism = u8::decode(bytes)?;
         match (version, dp_mechanism) {
             (.., DP_MECHANISM_NONE) => {
-                decode_u16_item_for_version::<()>(*version, bytes)?;
+                taskprov_decode_u16_prefixed::<()>(*version, bytes, |_version, inner| {
+                    <()>::decode(inner)
+                })?;
                 Ok(Self::None)
             }
             (DapVersion::DraftLatest, ..) => Ok(Self::NotImplemented {
@@ -207,18 +251,19 @@ impl ParameterizedEncode<DapVersion> for QueryConfig {
         match &self.var {
             QueryConfigVar::TimeInterval => {
                 QUERY_TYPE_TIME_INTERVAL.encode(bytes);
-                encode_u16_item_for_version(bytes, *version, &());
+                taskprov_encode_u16_prefixed(*version, bytes, |_, _| ());
             }
             QueryConfigVar::FixedSize { max_batch_size } => {
                 QUERY_TYPE_FIXED_SIZE.encode(bytes);
-                encode_u16_item_for_version(bytes, *version, max_batch_size);
+                taskprov_encode_u16_prefixed(*version, bytes, |_version, inner| {
+                    max_batch_size.encode(inner);
+                });
             }
             QueryConfigVar::NotImplemented { typ, param } => {
                 typ.encode(bytes);
-                match version {
-                    DapVersion::DraftLatest => encode_u16_bytes(bytes, param),
-                    DapVersion::Draft02 => bytes.extend_from_slice(param),
-                }
+                taskprov_encode_u16_prefixed(*version, bytes, |_version, inner| {
+                    inner.extend_from_slice(param);
+                });
             }
         }
     }
@@ -239,12 +284,18 @@ impl ParameterizedDecode<DapVersion> for QueryConfig {
         let query_type = query_type.unwrap_or(u8::decode(bytes)?);
         let var = match (version, query_type) {
             (.., QUERY_TYPE_TIME_INTERVAL) => {
-                decode_u16_item_for_version::<()>(*version, bytes)?;
+                taskprov_decode_u16_prefixed::<()>(*version, bytes, |_version, inner| {
+                    <()>::decode(inner)
+                })?;
                 QueryConfigVar::TimeInterval
             }
-            (.., QUERY_TYPE_FIXED_SIZE) => QueryConfigVar::FixedSize {
-                max_batch_size: decode_u16_item_for_version(*version, bytes)?,
-            },
+            (.., QUERY_TYPE_FIXED_SIZE) => {
+                taskprov_decode_u16_prefixed(*version, bytes, |_version, inner| {
+                    Ok(QueryConfigVar::FixedSize {
+                        max_batch_size: u32::decode(inner)?,
+                    })
+                })?
+            }
             (DapVersion::DraftLatest, ..) => QueryConfigVar::NotImplemented {
                 typ: query_type,
                 param: decode_u16_bytes(bytes)?,
@@ -318,6 +369,30 @@ impl ParameterizedDecode<DapVersion> for TaskConfig {
     }
 }
 
+fn taskprov_encode_u16_prefixed(
+    version: DapVersion,
+    bytes: &mut Vec<u8>,
+    e: impl Fn(DapVersion, &mut Vec<u8>),
+) {
+    match version {
+        DapVersion::DraftLatest => encode_u16_prefixed(version, bytes, e),
+        // draft02 compatibility: No length prefix is used.
+        DapVersion::Draft02 => e(version, bytes),
+    }
+}
+
+fn taskprov_decode_u16_prefixed<O>(
+    version: DapVersion,
+    bytes: &mut Cursor<&[u8]>,
+    d: impl Fn(DapVersion, &mut Cursor<&[u8]>) -> Result<O, CodecError>,
+) -> Result<O, CodecError> {
+    match version {
+        DapVersion::DraftLatest => decode_u16_prefixed(version, bytes, d),
+        // draft02 compatibility: No length prefix is used.
+        DapVersion::Draft02 => d(version, bytes),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use crate::test_versions;
@@ -443,7 +518,7 @@ mod tests {
         .is_err());
     }
 
-    fn roundtrip_vdaf_config(version: DapVersion) {
+    fn roundtrip_vdaf_config_prio2(version: DapVersion) {
         let vdaf_config = VdafConfig {
             dp_config: DpConfig::None,
             var: VdafTypeVar::Prio2 { dimension: 1337 },
@@ -458,7 +533,31 @@ mod tests {
         );
     }
 
-    test_versions! { roundtrip_vdaf_config }
+    test_versions! { roundtrip_vdaf_config_prio2 }
+
+    fn roundtrip_vdaf_config_prio3_sum_vec_field64_multiproof_hmac_sha256_aes128(
+        version: DapVersion,
+    ) {
+        let vdaf_config = VdafConfig {
+            dp_config: DpConfig::None,
+            var: VdafTypeVar::Prio3SumVecField64MultiproofHmacSha256Aes128 {
+                bits: 23,
+                length: 1337,
+                chunk_length: 42,
+                num_proofs: 99,
+            },
+        };
+        assert_eq!(
+            VdafConfig::get_decoded_with_param(
+                &version,
+                &vdaf_config.get_encoded_with_param(&version)
+            )
+            .unwrap(),
+            vdaf_config
+        );
+    }
+
+    test_versions! { roundtrip_vdaf_config_prio3_sum_vec_field64_multiproof_hmac_sha256_aes128 }
 
     #[test]
     fn roundtrip_vdaf_config_not_implemented_draft09() {
diff --git a/daphne/src/roles/aggregator.rs b/daphne/src/roles/aggregator.rs
index 817db7f..bf7a08a 100644
--- a/daphne/src/roles/aggregator.rs
+++ b/daphne/src/roles/aggregator.rs
@@ -25,9 +25,8 @@ use crate::{
 /// job.
 #[async_trait(?Send)]
 pub trait DapReportInitializer {
-    /// Initialize a sequence of reports that are in the "consumed" state by performing the early
-    /// validation steps (check if the report was replayed, belongs to a batch that has been
-    /// collected) and initializing VDAF preparation.
+    /// Initialize a sequence of reports that are in the "consumed" state by initializing VDAF
+    /// preparation.
     async fn initialize_reports<'req>(
         &self,
         is_leader: bool,
@@ -38,6 +37,13 @@ pub trait DapReportInitializer {
     ) -> Result<Vec<EarlyReportStateInitialized<'req>>, DapError>;
 }
 
+#[derive(Debug)]
+pub enum MergeAggShareError {
+    AlreadyCollected,
+    ReplaysDetected(HashSet<ReportId>),
+    Other(DapError),
+}
+
 /// DAP Aggregator functionality.
 #[async_trait(?Send)]
 pub trait DapAggregator<S>: HpkeDecrypter + DapReportInitializer + Sized {
@@ -117,15 +123,18 @@ pub trait DapAggregator<S>: HpkeDecrypter + DapReportInitializer + Sized {
     ///
     /// A span with the same buckets as the input `agg_share_span` where the value is one of 3
     /// possible sets of values:
-    /// - `Ok(None)` if all went well and no reports were replays.
-    /// - `Ok(Some(set))` if at least one report was a replay. This also means no aggregate shares where merged.
-    /// - `Err(err)` if an error occurred.
+    /// - `Ok(())` if all went well and no reports were replays.
+    /// - `Err(MergeAggShareError::ReplaysDetected)` if at least one report was a replay. This also
+    ///                                              means no aggregate shares where merged.
+    /// - `Err(MergeAggShareError::AlreadyCollected)` This span belong to an aggregate share that
+    ///                                               has been collected.
+    /// - `Err(MergeAggShareError::Other)` if another unrecoverable error occurred.
     async fn try_put_agg_share_span(
         &self,
         task_id: &TaskId,
         task_config: &DapTaskConfig,
         agg_share_span: DapAggregateSpan<DapAggregateShare>,
-    ) -> DapAggregateSpan<Result<HashSet<ReportId>, DapError>>;
+    ) -> DapAggregateSpan<Result<(), MergeAggShareError>>;
 
     /// Fetch the aggregate share for the given batch.
     async fn get_agg_share(
diff --git a/daphne/src/roles/helper.rs b/daphne/src/roles/helper.rs
index a90432a..1c61721 100644
--- a/daphne/src/roles/helper.rs
+++ b/daphne/src/roles/helper.rs
@@ -19,6 +19,7 @@ use crate::{
         ReportId, TaskId, TransitionFailure, TransitionVar,
     },
     metrics::{DaphneMetrics, DaphneRequestType},
+    roles::aggregator::MergeAggShareError,
     vdaf::ReportProcessedStatus,
     DapAggregateShare, DapAggregateSpan, DapAggregationJobState, DapError,
     DapHelperAggregationJobTransition, DapRequest, DapResource, DapResponse, DapTaskConfig,
@@ -266,12 +267,6 @@ pub trait DapHelper<S>: DapAggregator<S> {
             })
             .await?;
 
-        for transition in &agg_job_resp.transitions {
-            if let TransitionVar::Failed(failure) = &transition.var {
-                metrics.report_inc_by(&format!("rejected_{failure}"), 1);
-            }
-        }
-
         let out_shares_count = agg_job_resp
             .transitions
             .iter()
@@ -503,7 +498,7 @@ async fn finish_agg_job_and_aggregate<S>(
         for (_bucket, result) in put_shares_result {
             match result {
                 // This bucket had no replays.
-                (Ok(replays), reports) if replays.is_empty() => {
+                (Ok(()), reports) => {
                     // Every report in the bucket has been committed to aggregate storage.
                     report_status.extend(
                         reports.into_iter().map(|(report_id, _time)| {
@@ -512,7 +507,7 @@ async fn finish_agg_job_and_aggregate<S>(
                     );
                 }
                 // This bucket had replays.
-                (Ok(replays), _reports) => {
+                (Err(MergeAggShareError::ReplaysDetected(replays)), _reports) => {
                     // At least one report was replayed (no change to aggregate storage).
                     report_status.extend(replays.into_iter().map(|report_id| {
                         (
@@ -522,6 +517,16 @@ async fn finish_agg_job_and_aggregate<S>(
                     }));
                     inc_restart_metric.call_once(|| metrics.agg_job_put_span_retry_inc());
                 }
+                // This bucket is contained by an aggregate share span that has been collected.
+                (Err(MergeAggShareError::AlreadyCollected), reports) => {
+                    report_status.extend(reports.into_iter().map(|(report_id, _)| {
+                        (
+                            report_id,
+                            ReportProcessedStatus::Rejected(TransitionFailure::BatchCollected),
+                        )
+                    }));
+                    inc_restart_metric.call_once(|| metrics.agg_job_put_span_retry_inc());
+                }
                 // If this happens, the leader and helper can possibly have inconsistent state.
                 // The leader will still think all of the reports in this job have yet to be
                 // aggregated. But we could have aggregated some and not others due to the
@@ -531,7 +536,7 @@ async fn finish_agg_job_and_aggregate<S>(
                 // and if this error doesn't manifest itself all reports will be successfully
                 // aggregated. Which means that no reports will be lost in a such a state that
                 // they can never be aggregated.
-                (Err(e), _) => return Err(e),
+                (Err(MergeAggShareError::Other(other)), _) => return Err(other),
             }
         }
         if !inc_restart_metric.is_completed() {
@@ -544,6 +549,12 @@ async fn finish_agg_job_and_aggregate<S>(
                 .expect("usize to fit in u64");
             metrics.report_inc_by("aggregated", out_shares_count);
 
+            for transition in &agg_job_resp.transitions {
+                if let TransitionVar::Failed(failure) = &transition.var {
+                    metrics.report_inc_by(&format!("rejected_{failure}"), 1);
+                }
+            }
+
             return Ok(agg_job_resp);
         }
     }
@@ -586,7 +597,7 @@ mod tests {
             .map(|r| r.report_metadata.id)
             .collect::<Vec<_>>();
 
-        let req = test
+        let (_, req) = test
             .gen_test_agg_job_init_req(&task_id, DapVersion::Draft02, reports)
             .await;
 
diff --git a/daphne/src/roles/leader.rs b/daphne/src/roles/leader.rs
index 22249da..e609e19 100644
--- a/daphne/src/roles/leader.rs
+++ b/daphne/src/roles/leader.rs
@@ -8,7 +8,10 @@ use prio::codec::{Decode, ParameterizedDecode, ParameterizedEncode};
 use tracing::{debug, error};
 use url::Url;
 
-use super::{check_batch, check_request_content_type, resolve_taskprov, DapAggregator};
+use super::{
+    aggregator::MergeAggShareError, check_batch, check_request_content_type, resolve_taskprov,
+    DapAggregator,
+};
 use crate::{
     constants::DapMediaType,
     error::DapAbort,
@@ -333,6 +336,7 @@ pub trait DapLeader<S>: DapAuthorizedSender<S> + DapAggregator<S> {
                 metrics,
             )
             .await?;
+
         let (state, agg_job_init_req) = match transition {
             DapLeaderAggregationJobTransition::Continued(state, agg_job_init_req) => {
                 (state, agg_job_init_req)
@@ -437,26 +441,39 @@ pub trait DapLeader<S>: DapAuthorizedSender<S> + DapAggregator<S> {
 
         let out_shares_count = agg_span.report_count() as u64;
 
-        // At this point we're committed to aggregating the reports: if we do detect a report was
-        // replayed at this stage, then we may end up with a batch mismatch. However, this should
-        // only happen if there are multiple aggregation jobs in-flight that include the same
-        // report.
-        let replayed = self
+        // At this point we're committed to aggregating the reports: if we do detect an error (a
+        // report was replayed at this stage or the span overlaps with a collected batch), then we
+        // may end up with a batch mismatch. However, this should only happen if there are multiple
+        // aggregation jobs in-flight that include the same report.
+        let (replayed, collected) = self
             .try_put_agg_share_span(task_id, task_config, agg_span)
             .await
             .into_iter()
-            .map(|(_bucket, (result, _report_metadata))| {
-                result.map(|replayed_reports| replayed_reports.len())
+            .map(|(_bucket, (result, _report_metadata))| match result {
+                Ok(()) => Ok((0, 0)),
+                Err(MergeAggShareError::AlreadyCollected) => Ok((0, 1)),
+                Err(MergeAggShareError::ReplaysDetected(replays)) => Ok((replays.len(), 0)),
+                Err(MergeAggShareError::Other(e)) => Err(e),
             })
-            .sum::<Result<usize, _>>()?;
+            .try_fold((0, 0), |(replayed, collected), rc| {
+                let (r, c) = rc?;
+                Ok::<_, DapError>((replayed + r, collected + c))
+            })?;
 
         if replayed > 0 {
-            tracing::warn!(
+            tracing::error!(
                 replay_count = replayed,
                 "tried to aggregate replayed reports"
             );
         }
 
+        if collected > 0 {
+            tracing::error!(
+                collected_count = collected,
+                "tried to aggregate reports belonging to collected spans"
+            );
+        }
+
         metrics.report_inc_by("aggregated", out_shares_count);
         Ok(out_shares_count)
     }
diff --git a/daphne/src/roles/mod.rs b/daphne/src/roles/mod.rs
index fdba133..c5eb44c 100644
--- a/daphne/src/roles/mod.rs
+++ b/daphne/src/roles/mod.rs
@@ -3,7 +3,7 @@
 
 //! Trait definitions for Daphne backends.
 
-mod aggregator;
+pub mod aggregator;
 mod helper;
 mod leader;
 
@@ -170,14 +170,14 @@ mod test {
         test_versions,
         testing::{AggStore, MockAggregator, MockAggregatorReportSelector},
         vdaf::VdafVerifyKey,
-        DapAbort, DapAggregateShare, DapBatchBucket, DapCollectJob, DapError, DapGlobalConfig,
-        DapLeaderAggregationJobTransition, DapMeasurement, DapQueryConfig, DapRequest, DapResource,
-        DapTaskConfig, DapTaskParameters, DapVersion, MetaAggregationJobId, Prio3Config,
-        VdafConfig,
+        DapAbort, DapAggregateShare, DapAggregationJobState, DapBatchBucket, DapCollectJob,
+        DapError, DapGlobalConfig, DapLeaderAggregationJobTransition, DapMeasurement,
+        DapQueryConfig, DapRequest, DapResource, DapTaskConfig, DapTaskParameters, DapVersion,
+        MetaAggregationJobId, Prio3Config, VdafConfig,
     };
     use assert_matches::assert_matches;
     use matchit::Router;
-    use prio::codec::{Decode, ParameterizedEncode};
+    use prio::codec::{Decode, ParameterizedDecode, ParameterizedEncode};
     use rand::{thread_rng, Rng};
     use std::{collections::HashMap, sync::Arc, time::SystemTime, vec};
     use url::Url;
@@ -498,7 +498,7 @@ mod test {
             task_id: &TaskId,
             version: DapVersion,
             reports: Vec<Report>,
-        ) -> DapRequest<BearerToken> {
+        ) -> (DapAggregationJobState, DapRequest<BearerToken>) {
             let mut rng = thread_rng();
             let task_config = self.leader.unchecked_get_task_config(task_id).await;
             let part_batch_sel = match task_config.query {
@@ -510,7 +510,7 @@ mod test {
 
             let agg_job_id = MetaAggregationJobId::gen_for_version(version);
 
-            let DapLeaderAggregationJobTransition::Continued(_leader_state, agg_job_init_req) =
+            let DapLeaderAggregationJobTransition::Continued(leader_state, agg_job_init_req) =
                 task_config
                     .vdaf
                     .produce_agg_job_init_req(
@@ -529,15 +529,18 @@ mod test {
                 panic!("unexpected transition");
             };
 
-            self.leader_authorized_req(
-                task_id,
-                &task_config,
-                Some(&agg_job_id),
-                DapMediaType::AggregationJobInitReq,
-                agg_job_init_req,
-                task_config.helper_url.join("aggregate").unwrap(),
+            (
+                leader_state,
+                self.leader_authorized_req(
+                    task_id,
+                    &task_config,
+                    Some(&agg_job_id),
+                    DapMediaType::AggregationJobInitReq,
+                    agg_job_init_req,
+                    task_config.helper_url.join("aggregate").unwrap(),
+                )
+                .await,
             )
-            .await
         }
 
         pub async fn gen_test_agg_job_cont_req_with_round(
@@ -819,7 +822,7 @@ mod test {
     async fn handle_agg_job_init_req_unauthorized_request(version: DapVersion) {
         let t = Test::new(version);
         let report = t.gen_test_report(&t.time_interval_task_id).await;
-        let mut req = t
+        let (_, mut req) = t
             .gen_test_agg_job_init_req(&t.time_interval_task_id, version, vec![report])
             .await;
         req.sender_auth = None;
@@ -1058,7 +1061,7 @@ mod test {
 
         let mut report = t.gen_test_report(task_id).await;
         report.encrypted_input_shares[1].payload[0] ^= 0xff; // Cause decryption to fail
-        let req = t
+        let (_, req) = t
             .gen_test_agg_job_init_req(task_id, version, vec![report])
             .await;
 
@@ -1083,7 +1086,7 @@ mod test {
         let task_id = &t.time_interval_task_id;
 
         let report = t.gen_test_report(task_id).await;
-        let req = t
+        let (_, req) = t
             .gen_test_agg_job_init_req(task_id, version, vec![report])
             .await;
 
@@ -1105,7 +1108,7 @@ mod test {
         let task_id = &t.time_interval_task_id;
 
         let report = t.gen_test_report(task_id).await;
-        let req = t
+        let (leader_state, req) = t
             .gen_test_agg_job_init_req(task_id, version, vec![report.clone()])
             .await;
 
@@ -1126,17 +1129,57 @@ mod test {
             &t.helper.handle_agg_job_req(&req).await.unwrap().payload,
         )
         .unwrap();
-        let transition = &agg_job_resp.transitions[0];
+        let transitions = if version == DapVersion::Draft02 {
+            // in version 2 replays are only detected later.
+            let agg_job_id =
+                AggregationJobInitReq::get_decoded_with_param(&DapVersion::Draft02, &req.payload)
+                    .unwrap()
+                    .draft02_agg_job_id
+                    .unwrap();
+            let task_config = t.leader.unchecked_get_task_config(task_id).await;
+            let transition = task_config
+                .vdaf
+                .handle_agg_job_resp(
+                    task_id,
+                    &task_config,
+                    &MetaAggregationJobId::Draft02(agg_job_id),
+                    leader_state,
+                    agg_job_resp,
+                    t.leader.metrics(),
+                )
+                .unwrap();
+            let DapLeaderAggregationJobTransition::Uncommitted(
+                _,
+                AggregationJobContinueReq { transitions, .. },
+            ) = transition
+            else {
+                panic!("expected uncommitted transition, was {transition:?}");
+            };
+            let req = t
+                .gen_test_agg_job_cont_req(
+                    &MetaAggregationJobId::Draft02(agg_job_id),
+                    transitions,
+                    version,
+                )
+                .await;
+            AggregationJobResp::get_decoded(
+                &t.helper.handle_agg_job_req(&req).await.unwrap().payload,
+            )
+            .unwrap()
+            .transitions
+        } else {
+            agg_job_resp.transitions
+        };
 
         // Expect failure due to report store marked as collected.
         assert_matches!(
-            transition.var,
+            transitions[0].var,
             TransitionVar::Failed(TransitionFailure::ReportReplayed)
         );
 
         assert_metrics_include!(t.helper_registry, {
             r#"report_counter{env="test_helper",host="helper.org",status="rejected_report_replayed"}"#: 1,
-            r#"inbound_request_counter{env="test_helper",host="helper.org",type="aggregate"}"#: 1,
+            r#"inbound_request_counter{env="test_helper",host="helper.org",type="aggregate"}"#: if version == DapVersion::Draft02 { 2 } else { 1 },
             r#"aggregation_job_counter{env="test_helper",host="helper.org",status="started"}"#: 1,
         });
     }
@@ -1149,7 +1192,7 @@ mod test {
         let task_config = t.helper.unchecked_get_task_config(task_id).await;
 
         let report = t.gen_test_report(task_id).await;
-        let req = t
+        let (_, req) = t
             .gen_test_agg_job_init_req(task_id, version, vec![report])
             .await;
 
@@ -1204,7 +1247,7 @@ mod test {
         let task_id = &t.time_interval_task_id;
 
         let report = t.gen_test_report(task_id).await;
-        let req = t
+        let (_, req) = t
             .gen_test_agg_job_init_req(task_id, DapVersion::Draft02, vec![report])
             .await;
 
@@ -1752,7 +1795,11 @@ mod test {
 
     async_test_versions! { e2e_fixed_size }
 
-    async fn e2e_taskprov(version: DapVersion) {
+    async fn e2e_taskprov(
+        version: DapVersion,
+        vdaf_config: VdafConfig,
+        test_measurement: DapMeasurement,
+    ) {
         let t = Test::new(version);
 
         let (task_config, task_id, taskprov_advertisement, taskprov_report_extension_payload) =
@@ -1760,6 +1807,7 @@ mod test {
                 version,
                 min_batch_size: 1,
                 query: DapQueryConfig::FixedSize { max_batch_size: 2 },
+                vdaf: vdaf_config,
                 ..Default::default()
             }
             .to_config_with_taskprov(
@@ -1792,7 +1840,7 @@ mod test {
                     &hpke_config_list,
                     t.now,
                     &task_id,
-                    DapMeasurement::U32Vec(vec![1; 10]),
+                    test_measurement.clone(),
                     vec![Extension::Taskprov {
                         draft02_payload: match version {
                             DapVersion::DraftLatest => None,
@@ -1850,7 +1898,31 @@ mod test {
         });
     }
 
-    async_test_versions! { e2e_taskprov }
+    async fn e2e_taskprov_prio2(version: DapVersion) {
+        e2e_taskprov(
+            version,
+            VdafConfig::Prio2 { dimension: 10 },
+            DapMeasurement::U32Vec(vec![1; 10]),
+        )
+        .await;
+    }
+
+    async_test_versions! { e2e_taskprov_prio2 }
+
+    #[tokio::test]
+    async fn e2e_taskprov_prio3_sum_vec_field64_multiproof_hmac_sha256_aes128_draft09() {
+        e2e_taskprov(
+            DapVersion::DraftLatest,
+            VdafConfig::Prio3(Prio3Config::SumVecField64MultiproofHmacSha256Aes128 {
+                bits: 1,
+                length: 10,
+                chunk_length: 2,
+                num_proofs: 4,
+            }),
+            DapMeasurement::U64Vec(vec![1; 10]),
+        )
+        .await;
+    }
 
     fn early_metadata_checks(version: DapVersion) {
         let t = Test::new(version);
diff --git a/daphne/src/taskprov.rs b/daphne/src/taskprov.rs
index 1b1689d..1ed99ea 100644
--- a/daphne/src/taskprov.rs
+++ b/daphne/src/taskprov.rs
@@ -15,7 +15,7 @@ use crate::{
     },
     vdaf::VdafVerifyKey,
     DapAbort, DapError, DapQueryConfig, DapRequest, DapTaskConfig, DapTaskConfigMethod, DapVersion,
-    VdafConfig,
+    Prio3Config, VdafConfig,
 };
 use prio::codec::ParameterizedDecode;
 use ring::{
@@ -230,6 +230,32 @@ impl VdafConfig {
                     task_id: *task_id,
                 })?,
             }),
+            (
+                DapVersion::DraftLatest,
+                VdafTypeVar::Prio3SumVecField64MultiproofHmacSha256Aes128 {
+                    bits,
+                    length,
+                    chunk_length,
+                    num_proofs,
+                },
+            ) => Ok(VdafConfig::Prio3(
+                Prio3Config::SumVecField64MultiproofHmacSha256Aes128 {
+                    bits: bits.into(),
+                    length: length.try_into().map_err(|_| DapAbort::InvalidTask {
+                        detail: "length is larger than the system's word size".to_string(),
+                        task_id: *task_id,
+                    })?,
+                    chunk_length: chunk_length.try_into().map_err(|_| DapAbort::InvalidTask {
+                        detail: "chunk_length is larger than the system's word size".to_string(),
+                        task_id: *task_id,
+                    })?,
+                    num_proofs,
+                },
+            )),
+            (DapVersion::Draft02, var) => Err(DapAbort::InvalidTask {
+                detail: format!("draft02: unsupported VDAF: {var:?}"),
+                task_id: *task_id,
+            }),
             (.., VdafTypeVar::NotImplemented { typ, .. }) => Err(DapAbort::InvalidTask {
                 detail: format!("unimplemented VDAF type ({typ})"),
                 task_id: *task_id,
@@ -312,12 +338,30 @@ impl TryFrom<&VdafConfig> for messages::taskprov::VdafTypeVar {
     fn try_from(vdaf_config: &VdafConfig) -> Result<Self, DapError> {
         match vdaf_config {
             VdafConfig::Prio2 { dimension } => Ok(Self::Prio2 {
-                dimension: (*dimension)
-                    .try_into()
-                    .map_err(|_| fatal_error!(err = "Prio2 dimension is too large for taskprov"))?,
+                dimension: (*dimension).try_into().map_err(|_| {
+                    fatal_error!(err = "{vdaf_config}: dimension is too large for taskprov")
+                })?,
+            }),
+            VdafConfig::Prio3(Prio3Config::SumVecField64MultiproofHmacSha256Aes128 {
+                bits,
+                length,
+                chunk_length,
+                num_proofs,
+            }) => Ok(Self::Prio3SumVecField64MultiproofHmacSha256Aes128 {
+                bits: (*bits).try_into().map_err(|_| {
+                    fatal_error!(err = format!("{vdaf_config}: bits is too large for taskprov"))
+                })?,
+                length: (*length).try_into().map_err(|_| {
+                    fatal_error!(err = format!("{vdaf_config}: bits is too large for taskprov"))
+                })?,
+
+                chunk_length: (*chunk_length).try_into().map_err(|_| {
+                    fatal_error!(err = format!("{vdaf_config}: bits is too large for taskprov"))
+                })?,
+                num_proofs: *num_proofs,
             }),
-            VdafConfig::Prio3 { .. } => Err(fatal_error!(
-                err = "Prio3 is not currently supported for taskprov"
+            VdafConfig::Prio3(..) => Err(fatal_error!(
+                err = format!("{vdaf_config} is not currently supported for taskprov")
             )),
         }
     }
diff --git a/daphne/src/vdaf/mod.rs b/daphne/src/vdaf/mod.rs
index 8a09ddc..895ca6c 100644
--- a/daphne/src/vdaf/mod.rs
+++ b/daphne/src/vdaf/mod.rs
@@ -1028,7 +1028,6 @@ impl VdafConfig {
                 report_status,
                 initialized_reports,
                 agg_job_init_req,
-                metrics,
             ),
         }
     }
@@ -1098,7 +1097,6 @@ impl VdafConfig {
         report_status: &HashMap<ReportId, ReportProcessedStatus>,
         initialized_reports: &[EarlyReportStateInitialized<'_>],
         agg_job_init_req: &AggregationJobInitReq,
-        metrics: &DaphneMetrics,
     ) -> Result<DapHelperAggregationJobTransition<AggregationJobResp>, DapError> {
         let num_reports = agg_job_init_req.prep_inits.len();
         let mut agg_span = DapAggregateSpan::default();
@@ -1170,7 +1168,6 @@ impl VdafConfig {
 
                             Err(VdafError::Codec(..) | VdafError::Vdaf(..)) => {
                                 let failure = TransitionFailure::VdafPrepError;
-                                metrics.report_inc_by(&format!("rejected_{failure}"), 1);
                                 TransitionVar::Failed(failure)
                             }
                         }
@@ -1179,10 +1176,7 @@ impl VdafConfig {
                     EarlyReportStateInitialized::Rejected {
                         metadata: _,
                         failure,
-                    } => {
-                        metrics.report_inc_by(&format!("rejected_{failure}"), 1);
-                        TransitionVar::Failed(*failure)
-                    }
+                    } => TransitionVar::Failed(*failure),
                 },
             };
 
@@ -2162,10 +2156,6 @@ mod test {
             agg_job_resp.transitions[0].var,
             TransitionVar::Failed(TransitionFailure::HpkeDecryptError)
         );
-
-        assert_metrics_include!(t.helper_registry, {
-            r#"report_counter{env="test_helper",host="helper.org",status="rejected_hpke_decrypt_error"}"#: 1,
-        });
     }
 
     async_test_versions! { handle_agg_job_init_req_hpke_decrypt_err }
@@ -2191,10 +2181,6 @@ mod test {
             agg_job_resp.transitions[0].var,
             TransitionVar::Failed(TransitionFailure::HpkeUnknownConfigId)
         );
-
-        assert_metrics_include!(t.helper_registry, {
-            r#"report_counter{env="test_helper",host="helper.org",status="rejected_hpke_unknown_config_id"}"#: 1,
-        });
     }
 
     async_test_versions! { handle_agg_job_init_req_hpke_unknown_config_id }
@@ -2245,10 +2231,6 @@ mod test {
             agg_job_resp.transitions[1].var,
             TransitionVar::Failed(TransitionFailure::VdafPrepError)
         );
-
-        assert_metrics_include!(t.helper_registry, {
-            r#"report_counter{env="test_helper",host="helper.org",status="rejected_vdaf_prep_error"}"#: 2,
-        });
     }
 
     async_test_versions! { handle_agg_job_init_req_vdaf_prep_error }
diff --git a/daphne/src/vdaf/prio3.rs b/daphne/src/vdaf/prio3.rs
index 5a5baba..0f85e93 100644
--- a/daphne/src/vdaf/prio3.rs
+++ b/daphne/src/vdaf/prio3.rs
@@ -4,6 +4,7 @@
 //! Parameters for the [Prio3 VDAF](https://datatracker.ietf.org/doc/draft-patton-cfrg-vdaf/).
 
 use crate::{
+    messages::taskprov::VDAF_TYPE_PRIO3_SUM_VEC_FIELD64_MULTIPROOF_HMAC_SHA256_AES128,
     vdaf::{xof::XofHmacSha256Aes128, VdafError, VdafVerifyKey},
     DapAggregateResult, DapMeasurement, Prio3Config, VdafAggregateShare, VdafPrepMessage,
     VdafPrepState,
@@ -41,7 +42,7 @@ fn new_prio3_sum_vec_field64_multiproof_hmac_sha256_aes128(
     Ok(Prio3::new(
         2,
         num_proofs,
-        0xFFFF_1003,
+        VDAF_TYPE_PRIO3_SUM_VEC_FIELD64_MULTIPROOF_HMAC_SHA256_AES128,
         SumVec::new(bits, length, chunk_length).map_err(vdaf::VdafError::from)?,
     )?)
 }
diff --git a/daphne_worker/src/config.rs b/daphne_worker/src/config.rs
index 631184b..8c06093 100644
--- a/daphne_worker/src/config.rs
+++ b/daphne_worker/src/config.rs
@@ -121,10 +121,6 @@ pub(crate) struct DaphneWorkerConfig {
     /// configured by the Leader.
     pub(crate) helper_state_store_garbage_collect_after_secs: Option<Duration>,
 
-    /// Additional time to wait before deletng an instance of ReportsProcessed. Added to the value
-    /// of the `report_storage_epoch_duration` field of the global DAP configuration.
-    pub(crate) processed_alarm_safety_interval: Duration,
-
     /// Metrics push configuration.
     metrics_push_config: Option<MetricsPushConfig>,
 }
@@ -281,17 +277,6 @@ impl DaphneWorkerConfig {
             None
         };
 
-        let processed_alarm_safety_interval = Duration::from_secs(
-            env.var("DAP_PROCESSED_ALARM_SAFETY_INTERVAL")?
-                .to_string()
-                .parse()
-                .map_err(|err| {
-                    worker::Error::RustError(format!(
-                        "Failed to parse DAP_PROCESSED_ALARM_SAFETY_INTERVAL: {err}"
-                    ))
-                })?,
-        );
-
         const DAP_METRICS_PUSH_SERVER_URL: &str = "DAP_METRICS_PUSH_SERVER_URL";
         const DAP_METRICS_PUSH_BEARER_TOKEN: &str = "DAP_METRICS_PUSH_BEARER_TOKEN";
         let metrics_push_config = match (
@@ -331,7 +316,6 @@ impl DaphneWorkerConfig {
             taskprov,
             default_version,
             helper_state_store_garbage_collect_after_secs,
-            processed_alarm_safety_interval,
             metrics_push_config,
         })
     }
diff --git a/daphne_worker/src/durable/aggregate_store.rs b/daphne_worker/src/durable/aggregate_store.rs
index ea725ad..6720249 100644
--- a/daphne_worker/src/durable/aggregate_store.rs
+++ b/daphne_worker/src/durable/aggregate_store.rs
@@ -78,6 +78,9 @@ const MAX_CHUNK_SIZE: usize = 128_000;
 /// Key used to store metadata under.
 const METADATA_KEY: &str = "meta";
 
+/// Key used to store where this share has been collected
+const COLLECTED_KEY: &str = "collected";
+
 #[derive(Debug, Serialize, Deserialize, Clone, Copy)]
 #[serde(rename_all = "snake_case")]
 enum VdafKind {
@@ -168,7 +171,7 @@ impl AggregateStore {
             return Ok(DapAggregateShare::default());
         }
 
-        let meta_key = JsValue::from_str("meta");
+        let meta_key = JsValue::from_str(METADATA_KEY);
         let meta =
             serde_wasm_bindgen::from_value::<DapAggregateShareMetadata>(values.get(&meta_key))
                 .unwrap_or_else(|e| {
@@ -266,6 +269,13 @@ fn shard_bytes_to_object(
     Ok(())
 }
 
+#[derive(Debug, Serialize, Deserialize)]
+pub enum AggregateStoreMergeResp {
+    Ok,
+    ReplaysDetected(HashSet<ReportId>),
+    AlreadyCollected,
+}
+
 #[durable_object]
 impl DurableObject for AggregateStore {
     fn new(state: State, env: Env) -> Self {
@@ -288,6 +298,16 @@ impl DurableObject for AggregateStore {
 }
 
 impl AggregateStore {
+    async fn is_collected(&mut self) -> Result<bool> {
+        Ok(if let Some(collected) = self.collected {
+            collected
+        } else {
+            let collected = state_get_or_default(&self.state, COLLECTED_KEY).await?;
+            self.collected = Some(collected);
+            collected
+        })
+    }
+
     async fn handle(&mut self, req: Request) -> Result<Response> {
         let mut req = match self
             .schedule_for_garbage_collection(req, BINDING_DAP_AGGREGATE_STORE)
@@ -315,15 +335,22 @@ impl AggregateStore {
 
                 let chunks_map = js_sys::Object::default();
 
+                if self.is_collected().await? {
+                    return Response::from_json(&AggregateStoreMergeResp::AlreadyCollected);
+                }
+
                 {
                     // check for replays
                     let mut merged_report_ids = self.load_aggregated_report_ids().await?;
                     let repeat_ids = contained_reports
                         .iter()
                         .filter(|id| merged_report_ids.contains(id))
-                        .collect::<Vec<_>>();
+                        .copied()
+                        .collect::<HashSet<_>>();
                     if !repeat_ids.is_empty() {
-                        return Response::from_json(&repeat_ids);
+                        return Response::from_json(&AggregateStoreMergeResp::ReplaysDetected(
+                            repeat_ids,
+                        ));
                     }
                     merged_report_ids.extend(contained_reports);
                     let mut as_bytes =
@@ -353,7 +380,7 @@ impl AggregateStore {
 
                 self.state.storage().put_multiple_raw(chunks_map).await?;
 
-                Response::from_json::<[ReportId; 0]>(&[])
+                Response::from_json(&AggregateStoreMergeResp::Ok)
             }
 
             // Get the current aggregate share.
@@ -370,7 +397,7 @@ impl AggregateStore {
             // Non-idempotent (do not retry)
             // Output: `()`
             (DURABLE_AGGREGATE_STORE_MARK_COLLECTED, Method::Post) => {
-                self.state.storage().put("collected", true).await?;
+                self.state.storage().put(COLLECTED_KEY, true).await?;
                 self.collected = Some(true);
                 Response::from_json(&())
             }
@@ -380,14 +407,7 @@ impl AggregateStore {
             // Idempotent
             // Output: `bool`
             (DURABLE_AGGREGATE_STORE_CHECK_COLLECTED, Method::Get) => {
-                let collected = if let Some(collected) = self.collected {
-                    collected
-                } else {
-                    let collected = state_get_or_default(&self.state, "collected").await?;
-                    self.collected = Some(collected);
-                    collected
-                };
-                Response::from_json(&collected)
+                Response::from_json(&self.is_collected().await?)
             }
 
             _ => Err(int_err(format!(
diff --git a/daphne_worker/src/durable/mod.rs b/daphne_worker/src/durable/mod.rs
index d3ec279..5dc7d24 100644
--- a/daphne_worker/src/durable/mod.rs
+++ b/daphne_worker/src/durable/mod.rs
@@ -8,7 +8,6 @@ pub(crate) mod leader_agg_job_queue;
 pub(crate) mod leader_batch_queue;
 pub(crate) mod leader_col_job_queue;
 pub(crate) mod reports_pending;
-pub(crate) mod reports_processed;
 
 use crate::{
     int_err, now,
diff --git a/daphne_worker/src/durable/reports_processed.rs b/daphne_worker/src/durable/reports_processed.rs
deleted file mode 100644
index 1d5dd5a..0000000
--- a/daphne_worker/src/durable/reports_processed.rs
+++ /dev/null
@@ -1,317 +0,0 @@
-// Copyright (c) 2022 Cloudflare, Inc. All rights reserved.
-// SPDX-License-Identifier: BSD-3-Clause
-
-use crate::{
-    config::DaphneWorkerConfig,
-    durable::{create_span_from_request, state_get, BINDING_DAP_REPORTS_PROCESSED},
-    initialize_tracing, int_err,
-};
-use daphne::{
-    messages::{ReportId, ReportMetadata, TransitionFailure},
-    vdaf::{
-        EarlyReportState, EarlyReportStateConsumed, EarlyReportStateInitialized, VdafPrepMessage,
-        VdafPrepState, VdafVerifyKey,
-    },
-    DapError, VdafConfig,
-};
-use futures::{future::try_join_all, StreamExt, TryStreamExt};
-use prio::codec::{CodecError, ParameterizedDecode};
-use serde::{Deserialize, Serialize};
-use std::{borrow::Cow, collections::HashSet, future::ready, ops::ControlFlow, time::Duration};
-use tracing::Instrument;
-use worker::{
-    async_trait, durable_object, js_sys, wasm_bindgen, wasm_bindgen_futures, worker_sys, Env,
-    Method, Request, Response, Result, State,
-};
-
-use super::{req_parse, state_set_if_not_exists, Alarmed, DapDurableObject, GarbageCollectable};
-
-pub(crate) const DURABLE_REPORTS_PROCESSED_INITIALIZE: &str =
-    "/internal/do/reports_processed/initialize";
-pub(crate) const DURABLE_REPORTS_PROCESSED_INITIALIZED: &str =
-    "/internal/do/reports_processed/initialized";
-
-/// Durable Object (DO) for tracking which reports have been processed.
-///
-/// This object defines a single API endpoint, `DURABLE_REPORTS_PROCESSED_MARK_AGGREGATED`, which
-/// is used to mark a set of reports as aggregated. It returns the set of reports in that have
-/// already been aggregated (and thus need to be rejected by the caller).
-///
-/// The schema for stored report IDs is as follows:
-///
-/// ```text
-///     processed/<report_id> -> bool
-/// ```
-///
-/// where `<report_id>` is the hex-encoded report ID.
-#[durable_object]
-pub struct ReportsProcessed {
-    #[allow(dead_code)]
-    state: State,
-    env: Env,
-    config: DaphneWorkerConfig,
-    touched: bool,
-    alarmed: bool,
-}
-
-#[derive(Debug, Clone)]
-struct ReportIdKey<'s>(&'s ReportId, String);
-
-impl<'id> From<&'id ReportId> for ReportIdKey<'id> {
-    fn from(id: &'id ReportId) -> Self {
-        ReportIdKey(id, format!("processed/{}", id.to_hex()))
-    }
-}
-
-#[durable_object]
-impl DurableObject for ReportsProcessed {
-    fn new(state: State, env: Env) -> Self {
-        initialize_tracing(&env);
-        let config =
-            DaphneWorkerConfig::from_worker_env(&env).expect("failed to load configuration");
-        Self {
-            state,
-            env,
-            config,
-            touched: false,
-            alarmed: false,
-        }
-    }
-
-    async fn fetch(&mut self, req: Request) -> Result<Response> {
-        let span = create_span_from_request(&req);
-        self.handle(req).instrument(span).await
-    }
-
-    async fn alarm(&mut self) -> Result<Response> {
-        self.state.storage().delete_all().await?;
-        self.alarmed = false;
-        self.touched = false;
-        Response::from_json(&())
-    }
-}
-
-impl ReportsProcessed {
-    async fn handle(&mut self, req: Request) -> Result<Response> {
-        let mut req = match self
-            .schedule_for_garbage_collection(req, BINDING_DAP_REPORTS_PROCESSED)
-            .await?
-        {
-            ControlFlow::Continue(req) => req,
-            // This req was a GC request and as such we must return from this function.
-            ControlFlow::Break(_) => return Response::from_json(&()),
-        };
-
-        self.ensure_alarmed(
-            Duration::from_secs(self.config.global.report_storage_epoch_duration)
-                .saturating_add(self.config.processed_alarm_safety_interval),
-        )
-        .await?;
-
-        match (req.path().as_ref(), req.method()) {
-            (DURABLE_REPORTS_PROCESSED_INITIALIZED, Method::Post) => {
-                let to_mark = req_parse::<Vec<ReportId>>(&mut req).await?;
-                let state = &self.state;
-                let replays = futures::stream::iter(&to_mark)
-                    .map(|id| async move {
-                        state_set_if_not_exists(state, &format!("processed/{id}"), &true)
-                            .await
-                            .map(|o| o.is_some().then_some(id))
-                    })
-                    .buffer_unordered(usize::MAX)
-                    .try_filter_map(|replay| ready(Ok(replay)))
-                    .try_collect::<Vec<&ReportId>>()
-                    .await?;
-
-                Response::from_json(&replays)
-            }
-            // Initialize a report:
-            //  * Ensure the report wasn't replayed
-            //  * Ensure the report won't be included in a batch that was already collected
-            //  * Initialize VDAF preparation.
-            //
-            // Idempotent
-            // Input: `ReportsProcessedReq`
-            // Output: `ReportsProcessedResp`
-            (DURABLE_REPORTS_PROCESSED_INITIALIZE, Method::Post) => {
-                let reports_processed_request: ReportsProcessedReq = req_parse(&mut req).await?;
-                let result = try_join_all(
-                    reports_processed_request
-                        .consumed_reports
-                        .iter()
-                        .filter(|consumed_report| consumed_report.is_ready())
-                        .map(|consumed_report| async {
-                            if let Some(exists) = state_get::<bool>(
-                                &self.state,
-                                &format!("processed/{}", consumed_report.metadata().id.to_hex()),
-                            )
-                            .await?
-                            {
-                                if exists {
-                                    return Result::Ok(Some(consumed_report.metadata().id));
-                                }
-                            }
-                            Ok(None)
-                        }),
-                )
-                .await?;
-                let replayed_reports = result.into_iter().flatten().collect::<HashSet<ReportId>>();
-
-                let initialized_reports = reports_processed_request
-                    .consumed_reports
-                    .into_iter()
-                    .map(|consumed_report| {
-                        if replayed_reports.contains(&consumed_report.metadata().id) {
-                            Ok(consumed_report.into_initialized_rejected_due_to(
-                                TransitionFailure::ReportReplayed,
-                            ))
-                        } else {
-                            EarlyReportStateInitialized::initialize(
-                                reports_processed_request.is_leader,
-                                &reports_processed_request.vdaf_verify_key,
-                                &reports_processed_request.vdaf_config,
-                                consumed_report,
-                            )
-                        }
-                    })
-                    .collect::<std::result::Result<Vec<EarlyReportStateInitialized>, DapError>>()
-                    .map_err(|e| {
-                        int_err(format!(
-                            "ReportsProcessed: failed to initialize a report: {e}"
-                        ))
-                    })?;
-
-                Response::from_json(&ReportsProcessedResp {
-                    is_leader: reports_processed_request.is_leader,
-                    vdaf_config: reports_processed_request.vdaf_config,
-                    initialized_reports,
-                })
-            }
-
-            _ => Err(int_err(format!(
-                "ReportsProcessed: unexpected request: method={:?}; path={:?}",
-                req.method(),
-                req.path()
-            ))),
-        }
-    }
-}
-
-impl DapDurableObject for ReportsProcessed {
-    #[inline(always)]
-    fn state(&self) -> &State {
-        &self.state
-    }
-
-    #[inline(always)]
-    fn deployment(&self) -> crate::config::DaphneWorkerDeployment {
-        self.config.deployment
-    }
-}
-
-#[async_trait::async_trait]
-impl Alarmed for ReportsProcessed {
-    #[inline(always)]
-    fn alarmed(&mut self) -> &mut bool {
-        &mut self.alarmed
-    }
-}
-
-#[async_trait::async_trait(?Send)]
-impl GarbageCollectable for ReportsProcessed {
-    #[inline(always)]
-    fn touched(&mut self) -> &mut bool {
-        &mut self.touched
-    }
-
-    #[inline(always)]
-    fn env(&self) -> &Env {
-        &self.env
-    }
-}
-
-#[derive(Serialize, Deserialize)]
-pub(crate) struct ReportsProcessedReq<'req> {
-    pub(crate) is_leader: bool,
-    pub(crate) vdaf_verify_key: VdafVerifyKey,
-    pub(crate) vdaf_config: VdafConfig,
-    pub(crate) consumed_reports: Vec<EarlyReportStateConsumed<'req>>,
-}
-
-#[derive(Serialize, Deserialize)]
-#[serde(try_from = "ShadowReportsProcessedResp")]
-pub(crate) struct ReportsProcessedResp<'req> {
-    pub(crate) is_leader: bool,
-    pub(crate) vdaf_config: VdafConfig,
-    pub(crate) initialized_reports: Vec<EarlyReportStateInitialized<'req>>,
-}
-
-// we need this custom deserializer because VdafPrepState and VdafPrepMessage don't implement
-// Decode, only ParameterizedDecode.
-#[derive(Deserialize)]
-struct ShadowReportsProcessedResp {
-    pub(crate) is_leader: bool,
-    pub(crate) vdaf_config: VdafConfig,
-    pub(crate) initialized_reports: Vec<EarlyReportStateInitializedOwned>,
-}
-
-#[derive(Deserialize)]
-#[serde(rename_all = "snake_case")]
-pub enum EarlyReportStateInitializedOwned {
-    Ready {
-        metadata: ReportMetadata,
-        #[serde(with = "hex")]
-        public_share: Vec<u8>,
-        #[serde(with = "hex")]
-        state: Vec<u8>,
-        #[serde(with = "hex")]
-        message: Vec<u8>,
-    },
-    Rejected {
-        metadata: ReportMetadata,
-        failure: TransitionFailure,
-    },
-}
-
-impl TryFrom<ShadowReportsProcessedResp> for ReportsProcessedResp<'_> {
-    type Error = CodecError;
-
-    fn try_from(other: ShadowReportsProcessedResp) -> std::result::Result<Self, CodecError> {
-        let initialized_reports = other
-            .initialized_reports
-            .into_iter()
-            .map(|initialized_report| match initialized_report {
-                EarlyReportStateInitializedOwned::Ready {
-                    metadata,
-                    public_share,
-                    state,
-                    message,
-                } => {
-                    let state = VdafPrepState::get_decoded_with_param(
-                        &(&other.vdaf_config, other.is_leader),
-                        &state,
-                    )?;
-                    let message = VdafPrepMessage::get_decoded_with_param(&state, &message)?;
-
-                    Ok(EarlyReportStateInitialized::Ready {
-                        metadata: Cow::Owned(metadata),
-                        public_share: Cow::Owned(public_share),
-                        state,
-                        message,
-                    })
-                }
-                EarlyReportStateInitializedOwned::Rejected { metadata, failure } => {
-                    Ok(EarlyReportStateInitialized::Rejected {
-                        metadata: Cow::Owned(metadata),
-                        failure,
-                    })
-                }
-            })
-            .collect::<std::result::Result<Vec<EarlyReportStateInitialized>, CodecError>>()?;
-        Ok(Self {
-            is_leader: other.is_leader,
-            vdaf_config: other.vdaf_config,
-            initialized_reports,
-        })
-    }
-}
diff --git a/daphne_worker/src/roles/aggregator.rs b/daphne_worker/src/roles/aggregator.rs
index 2b68090..c53fe0c 100644
--- a/daphne_worker/src/roles/aggregator.rs
+++ b/daphne_worker/src/roles/aggregator.rs
@@ -8,13 +8,11 @@ use crate::{
     config::{DapTaskConfigKvPair, DaphneWorker},
     durable::{
         aggregate_store::{
-            AggregateStoreMergeReq, DURABLE_AGGREGATE_STORE_CHECK_COLLECTED,
-            DURABLE_AGGREGATE_STORE_GET, DURABLE_AGGREGATE_STORE_MARK_COLLECTED,
-            DURABLE_AGGREGATE_STORE_MERGE,
+            AggregateStoreMergeReq, AggregateStoreMergeResp,
+            DURABLE_AGGREGATE_STORE_CHECK_COLLECTED, DURABLE_AGGREGATE_STORE_GET,
+            DURABLE_AGGREGATE_STORE_MARK_COLLECTED, DURABLE_AGGREGATE_STORE_MERGE,
         },
-        durable_name_agg_store,
-        reports_processed::DURABLE_REPORTS_PROCESSED_INITIALIZED,
-        BINDING_DAP_AGGREGATE_STORE, BINDING_DAP_REPORTS_PROCESSED,
+        durable_name_agg_store, BINDING_DAP_AGGREGATE_STORE,
     },
     now,
 };
@@ -24,76 +22,29 @@ use daphne::{
     auth::BearerTokenProvider,
     fatal_error,
     hpke::HpkeConfig,
-    messages::{BatchId, BatchSelector, PartialBatchSelector, ReportId, TaskId, TransitionFailure},
+    messages::{BatchId, BatchSelector, PartialBatchSelector, TaskId, TransitionFailure},
     metrics::DaphneMetrics,
-    roles::{DapAggregator, DapReportInitializer},
+    roles::{aggregator::MergeAggShareError, DapAggregator, DapReportInitializer},
     vdaf::{EarlyReportState, EarlyReportStateConsumed, EarlyReportStateInitialized},
     DapAggregateShare, DapAggregateSpan, DapBatchBucket, DapError, DapGlobalConfig, DapRequest,
     DapSender, DapTaskConfig,
 };
-use futures::{future::try_join_all, StreamExt, TryFutureExt, TryStreamExt};
-use std::collections::{HashMap, HashSet};
+use futures::{future::try_join_all, StreamExt};
 
 #[async_trait(?Send)]
 impl DapReportInitializer for DaphneWorker<'_> {
     async fn initialize_reports<'req>(
         &self,
         is_leader: bool,
-        task_id: &TaskId,
+        _task_id: &TaskId,
         task_config: &DapTaskConfig,
-        part_batch_sel: &PartialBatchSelector,
+        _part_batch_sel: &PartialBatchSelector,
         consumed_reports: Vec<EarlyReportStateConsumed<'req>>,
     ) -> Result<Vec<EarlyReportStateInitialized<'req>>, DapError> {
-        let durable = self.durable();
-        let span = task_config
-            .as_ref()
-            .batch_span_for_meta(part_batch_sel, consumed_reports.iter())?;
-        let mut reports_processed_request_data = HashMap::new();
-        let collected_reports = {
-            let task_id_hex = task_id.to_hex();
-
-            let mut agg_store_request_names = Vec::new();
-            for (bucket, ((), report_ids_and_time)) in span.iter() {
-                for (id, time) in report_ids_and_time {
-                    let durable_name = self.config().durable_name_report_store(
-                        task_config.as_ref(),
-                        &task_id_hex,
-                        id,
-                        *time,
-                    );
-
-                    reports_processed_request_data
-                        .entry(durable_name)
-                        .or_insert_with(Vec::new)
-                        .push(id);
-                }
-                agg_store_request_names.push((
-                    bucket,
-                    durable_name_agg_store(task_config.version, &task_id_hex, bucket),
-                ));
-            }
-
-            // Send AggregateStore requests.
-            futures::stream::iter(agg_store_request_names)
-                .map(|(bucket, durable_name)| {
-                    durable
-                        .get(
-                            BINDING_DAP_AGGREGATE_STORE,
-                            DURABLE_AGGREGATE_STORE_CHECK_COLLECTED,
-                            durable_name,
-                        )
-                        .map_ok(move |collected| (bucket, collected))
-                })
-                .buffer_unordered(usize::MAX)
-                .try_collect::<HashMap<&DapBatchBucket, bool>>()
-                .map_err(|e| fatal_error!(err = ?e, "failed to check collected"))
-                .await?
-        };
-
         let min_time = self.least_valid_report_time(self.get_current_time());
         let max_time = self.greatest_valid_report_time(self.get_current_time());
 
-        let mut initialized_reports = consumed_reports
+        let initialized_reports = consumed_reports
             .into_iter()
             .map(|consumed_report| {
                 let metadata = consumed_report.metadata();
@@ -104,11 +55,6 @@ impl DapReportInitializer for DaphneWorker<'_> {
                 } else if metadata.time > max_time {
                     consumed_report
                         .into_initialized_rejected_due_to(TransitionFailure::ReportTooEarly)
-                } else if collected_reports
-                    [&task_config.bucket_for(part_batch_sel, &consumed_report)]
-                {
-                    consumed_report
-                        .into_initialized_rejected_due_to(TransitionFailure::BatchCollected)
                 } else {
                     EarlyReportStateInitialized::initialize(
                         is_leader,
@@ -123,31 +69,6 @@ impl DapReportInitializer for DaphneWorker<'_> {
             .collect::<Result<Vec<_>, DapError>>()
             .map_err(|e| fatal_error!(err = ?e, "failed to initialize a report"))?;
 
-        let replayed_reports_check = futures::stream::iter(reports_processed_request_data)
-            .map(|(durable_name, reports)| async {
-                durable
-                    .post::<_, HashSet<ReportId>>(
-                        BINDING_DAP_REPORTS_PROCESSED,
-                        DURABLE_REPORTS_PROCESSED_INITIALIZED,
-                        durable_name,
-                        reports,
-                    )
-                    .await
-            })
-            .buffer_unordered(usize::MAX)
-            .try_fold(HashSet::new(), |mut acc, replays| async {
-                acc.extend(replays);
-                Ok(acc)
-            })
-            .await
-            .map_err(|e| fatal_error!(err = ?e, "checking for replayed reports"))?;
-
-        for rep in &mut initialized_reports {
-            if replayed_reports_check.contains(&rep.metadata().id) {
-                rep.reject_due_to(TransitionFailure::ReportReplayed);
-            }
-        }
-
         Ok(initialized_reports)
     }
 }
@@ -386,7 +307,7 @@ impl<'srv> DapAggregator<DaphneWorkerAuth> for DaphneWorker<'srv> {
         task_id: &TaskId,
         task_config: &DapTaskConfig,
         agg_share_span: DapAggregateSpan<DapAggregateShare>,
-    ) -> DapAggregateSpan<Result<HashSet<ReportId>, DapError>> {
+    ) -> DapAggregateSpan<Result<(), MergeAggShareError>> {
         let task_id_hex = task_id.to_hex();
         let durable = self.durable().with_retry();
 
@@ -395,7 +316,7 @@ impl<'srv> DapAggregator<DaphneWorkerAuth> for DaphneWorker<'srv> {
                 let agg_store_name =
                     durable_name_agg_store(task_config.version, &task_id_hex, &bucket);
                 let result = durable
-                    .post::<_, HashSet<ReportId>>(
+                    .post::<_, AggregateStoreMergeResp>(
                         BINDING_DAP_AGGREGATE_STORE,
                         DURABLE_AGGREGATE_STORE_MERGE,
                         agg_store_name,
@@ -406,6 +327,16 @@ impl<'srv> DapAggregator<DaphneWorkerAuth> for DaphneWorker<'srv> {
                     )
                     .await
                     .map_err(|e| fatal_error!(err = ?e));
+                let result = match result {
+                    Ok(AggregateStoreMergeResp::Ok) => Ok(()),
+                    Ok(AggregateStoreMergeResp::AlreadyCollected) => {
+                        Err(MergeAggShareError::AlreadyCollected)
+                    }
+                    Ok(AggregateStoreMergeResp::ReplaysDetected(replays)) => {
+                        Err(MergeAggShareError::ReplaysDetected(replays))
+                    }
+                    Err(e) => Err(MergeAggShareError::Other(e)),
+                };
                 (bucket, (result, report_metadatas))
             })
             .buffer_unordered(usize::MAX)
diff --git a/docker/wrangler.toml b/docker/wrangler.toml
index 6f3c9e4..ce27352 100644
--- a/docker/wrangler.toml
+++ b/docker/wrangler.toml
@@ -36,7 +36,6 @@ DAP_GLOBAL_CONFIG = """{
      "supported_hpke_kems": ["x25519_hkdf_sha256"],
      "allow_taskprov": true
 }"""
-DAP_PROCESSED_ALARM_SAFETY_INTERVAL = "300"
 DAP_DEPLOYMENT = "dev"
 DAP_TASKPROV_HPKE_COLLECTOR_CONFIG = """{
   "id": 23,
@@ -65,7 +64,6 @@ bindings = [
     { name = "DAP_LEADER_COL_JOB_QUEUE", class_name = "LeaderCollectionJobQueue" },
     { name = "DAP_GARBAGE_COLLECTOR", class_name = "GarbageCollector" },
     { name = "DAP_REPORTS_PENDING", class_name = "ReportsPending" },
-    { name = "DAP_REPORTS_PROCESSED", class_name = "ReportsProcessed" },
 ]
 
 
@@ -116,7 +114,6 @@ DAP_GLOBAL_CONFIG = """{
   "supported_hpke_kems": ["x25519_hkdf_sha256"],
   "allow_taskprov": true
 }"""
-DAP_PROCESSED_ALARM_SAFETY_INTERVAL = "300"
 DAP_DEPLOYMENT = "dev"
 DAP_TASKPROV_HPKE_COLLECTOR_CONFIG = """{
   "id": 23,
@@ -139,7 +136,6 @@ bindings = [
     { name = "DAP_AGGREGATE_STORE", class_name = "AggregateStore" },
     { name = "DAP_HELPER_STATE_STORE", class_name = "HelperStateStore" },
     { name = "DAP_GARBAGE_COLLECTOR", class_name = "GarbageCollector" },
-    { name = "DAP_REPORTS_PROCESSED", class_name = "ReportsProcessed" },
 ]
 
 
@@ -166,5 +162,4 @@ new_classes = [
     "LeaderCollectionJobQueue",
     "GarbageCollector",
     "ReportsPending",
-    "ReportsProcessed",
 ]


Test patch:
diff --git a/daphne/src/testing.rs b/daphne/src/testing.rs
index 7ae30f1..dd360d6 100644
--- a/daphne/src/testing.rs
+++ b/daphne/src/testing.rs
@@ -16,7 +16,10 @@ use crate::{
         TaskId, Time, TransitionFailure,
     },
     metrics::DaphneMetrics,
-    roles::{DapAggregator, DapAuthorizedSender, DapHelper, DapLeader, DapReportInitializer},
+    roles::{
+        aggregator::MergeAggShareError, DapAggregator, DapAuthorizedSender, DapHelper, DapLeader,
+        DapReportInitializer,
+    },
     vdaf::{EarlyReportState, EarlyReportStateConsumed, EarlyReportStateInitialized},
     DapAbort, DapAggregateResult, DapAggregateShare, DapAggregateSpan, DapAggregationJobState,
     DapAggregationJobUncommitted, DapBatchBucket, DapCollectJob, DapError, DapGlobalConfig,
@@ -720,12 +723,10 @@ impl MockAggregator {
 
     /// Conducts checks on a received report to see whether:
     /// 1) the report falls into a batch that has been already collected, or
-    /// 2) the report has been submitted by the client in the past.
-    fn check_report_early_fail(
+    fn check_report_has_been_collected(
         &self,
         task_id: &TaskId,
         bucket: &DapBatchBucket,
-        id: &ReportId,
     ) -> Option<TransitionFailure> {
         // Check AggStateStore to see whether the report is part of a batch that has already
         // been collected.
@@ -735,16 +736,6 @@ impl MockAggregator {
             return Some(TransitionFailure::BatchCollected);
         }
 
-        // Check whether the same report has been submitted in the past.
-        let mut guard = self
-            .report_store
-            .lock()
-            .expect("report_store: failed to lock");
-        let report_store = guard.entry(*task_id).or_default();
-        if report_store.processed.contains(id) {
-            return Some(TransitionFailure::ReportReplayed);
-        }
-
         None
     }
 
@@ -948,7 +939,8 @@ impl DapReportInitializer for MockAggregator {
         for (bucket, ((), report_ids_and_time)) in span.iter() {
             for (id, _) in report_ids_and_time {
                 // Check whether Report has been collected or replayed.
-                if let Some(transition_failure) = self.check_report_early_fail(task_id, bucket, id)
+                if let Some(transition_failure) =
+                    self.check_report_has_been_collected(task_id, bucket)
                 {
                     early_fails.insert(*id, transition_failure);
                 };
@@ -1078,7 +1070,7 @@ impl DapAggregator<BearerToken> for MockAggregator {
         task_id: &TaskId,
         _task_config: &DapTaskConfig,
         agg_agg_span: DapAggregateSpan<DapAggregateShare>,
-    ) -> DapAggregateSpan<Result<HashSet<ReportId>, DapError>> {
+    ) -> DapAggregateSpan<Result<(), MergeAggShareError>> {
         let mut report_store_guard = self
             .report_store
             .lock()
@@ -1089,7 +1081,7 @@ impl DapAggregator<BearerToken> for MockAggregator {
 
         agg_agg_span
             .into_iter()
-            .map(|(bucket, (agg_share, report_metadatas))| {
+            .map(|(bucket, (agg_share_delta, report_metadatas))| {
                 let replayed = report_metadatas
                     .iter()
                     .map(|(id, _)| *id)
@@ -1101,14 +1093,17 @@ impl DapAggregator<BearerToken> for MockAggregator {
                         .processed
                         .extend(report_metadatas.iter().map(|(id, _)| *id));
                     // Add to aggregate share.
-                    agg_store
-                        .entry(bucket.clone())
-                        .or_default()
-                        .agg_share
-                        .merge(agg_share.clone())
-                        .map(|()| HashSet::new())
+                    let agg_share = agg_store.entry(bucket.clone()).or_default();
+                    if agg_share.collected {
+                        Err(MergeAggShareError::AlreadyCollected)
+                    } else {
+                        agg_share
+                            .agg_share
+                            .merge(agg_share_delta.clone())
+                            .map_err(MergeAggShareError::Other)
+                    }
                 } else {
-                    Ok(replayed)
+                    Err(MergeAggShareError::ReplaysDetected(replayed))
                 };
                 (bucket, (result, report_metadatas))
             })
@@ -1240,21 +1235,14 @@ impl DapLeader<BearerToken> for MockAggregator {
             .await
             .expect("could not determine batch for report");
 
-        // Check whether Report has been collected or replayed.
-        if let Some(transition_failure) =
-            self.check_report_early_fail(task_id, &bucket, &report.report_metadata.id)
-        {
-            return Err(DapError::Transition(transition_failure));
-        };
-
         // Store Report for future processing.
         let mut guard = self
             .report_store
             .lock()
             .expect("report_store: failed to lock");
         let queue = guard
-            .get_mut(task_id)
-            .expect("report_store: unrecognized task")
+            .entry(*task_id)
+            .or_default()
             .pending
             .entry(bucket)
             .or_default();
diff --git a/daphne_worker_test/wrangler.toml b/daphne_worker_test/wrangler.toml
index bb4b17d..6103c9f 100644
--- a/daphne_worker_test/wrangler.toml
+++ b/daphne_worker_test/wrangler.toml
@@ -36,7 +36,6 @@ DAP_GLOBAL_CONFIG = """{
      "supported_hpke_kems": ["x25519_hkdf_sha256"],
      "allow_taskprov": true
 }"""
-DAP_PROCESSED_ALARM_SAFETY_INTERVAL = "300"
 DAP_DEPLOYMENT = "dev"
 DAP_TASKPROV_HPKE_COLLECTOR_CONFIG = """{
   "id": 23,
@@ -65,7 +64,6 @@ bindings = [
     { name = "DAP_LEADER_COL_JOB_QUEUE", class_name = "LeaderCollectionJobQueue" },
     { name = "DAP_GARBAGE_COLLECTOR", class_name = "GarbageCollector" },
     { name = "DAP_REPORTS_PENDING", class_name = "ReportsPending" },
-    { name = "DAP_REPORTS_PROCESSED", class_name = "ReportsProcessed" },
 ]
 
 
@@ -116,7 +114,6 @@ DAP_GLOBAL_CONFIG = """{
   "supported_hpke_kems": ["x25519_hkdf_sha256"],
   "allow_taskprov": true
 }"""
-DAP_PROCESSED_ALARM_SAFETY_INTERVAL = "300"
 DAP_DEPLOYMENT = "dev"
 DAP_TASKPROV_HPKE_COLLECTOR_CONFIG = """{
   "id": 23,
@@ -139,7 +136,6 @@ bindings = [
     { name = "DAP_AGGREGATE_STORE", class_name = "AggregateStore" },
     { name = "DAP_HELPER_STATE_STORE", class_name = "HelperStateStore" },
     { name = "DAP_GARBAGE_COLLECTOR", class_name = "GarbageCollector" },
-    { name = "DAP_REPORTS_PROCESSED", class_name = "ReportsProcessed" },
 ]
 
 
@@ -166,5 +162,4 @@ new_classes = [
     "LeaderCollectionJobQueue",
     "GarbageCollector",
     "ReportsPending",
-    "ReportsProcessed",
 ]