Repository: cloudflare/daphne
Pull request #59 (2022-07-23 20:48:01)

daphne_worker: Process reports in order of arrival
Based on #58 (merge that first).
Partially addresses #25.

Change report selector for `DaphneWorkerConfig` so that instead of
scheduling work per task and batch interval, reports are processed in
order of arrival.

A new DO is implemented, `LeaderAggregationJobQueue`, which maintaines
a queue of `ReportStore` instances. Once a report arrrives to an empty
`ReportStore` instance, an "aggregation job" is added to the queue; once
the instance becomes empty, the job is removed from the queue. Jobs are
ordered by the first report's time of arrival so that the oldest data is
processed first.

Other changes:

* daphne: Remove `task_id` from `DapLeader::get_pending_collect_jobs()`
  so that scheduling work on collect jobs doesn't have to depend on a
  particular task.

* Remove a TODO that was addressed by a previous change.

cjpatton: Rebased on top of @nakatsuka-y's last PR.

Patch:
diff --git a/daphne/src/roles.rs b/daphne/src/roles.rs
index 3da4fd4..b5d95ad 100644
--- a/daphne/src/roles.rs
+++ b/daphne/src/roles.rs
@@ -166,11 +166,9 @@ pub trait DapLeader<S>: DapAuthorizedSender<S> + DapAggregator<S> {
         collect_id: &Id,
     ) -> Result<DapCollectJob, DapError>;
 
-    /// Return the sequence of pending collect jobs for a given task, in order of priority.
-    async fn get_pending_collect_jobs(
-        &self,
-        task_id: &Id,
-    ) -> Result<HashMap<Id, Vec<CollectReq>>, DapError>;
+    /// Fetch the current collect job queue. The result is the sequence of collect ID and request
+    /// pairs, in order of priority.
+    async fn get_pending_collect_jobs(&self) -> Result<Vec<(Id, CollectReq)>, DapError>;
 
     /// Complete a collect job by assigning it the completed [`CollectResp`](crate::messages::CollectResp).
     async fn finish_collect_job(
@@ -240,6 +238,11 @@ pub trait DapLeader<S>: DapAuthorizedSender<S> + DapAggregator<S> {
 
     /// Run the aggregation sub-protocol for the given set of reports. Return the number of reports
     /// that were aggregated successfully.
+    //
+    // TODO Handle non-encodable messages gracefully. The length of `reports` may be too long to
+    // encode in `AggregateInitializeReq`, in which case this method will panic. We should increase
+    // the capacity of this message in the spec. In the meantime, we should at a minimum log this
+    // when it happens.
     async fn run_agg_job(
         &self,
         task_id: &Id,
@@ -316,7 +319,6 @@ pub trait DapLeader<S>: DapAuthorizedSender<S> + DapAggregator<S> {
     async fn run_collect_job(
         &self,
         collect_id: &Id,
-        task_id: &Id,
         task_config: &DapTaskConfig,
         collect_req: CollectReq,
     ) -> Result<u64, DapAbort> {
@@ -349,7 +351,7 @@ pub trait DapLeader<S>: DapAuthorizedSender<S> + DapAggregator<S> {
         // Send AggregateShareReq and receive AggregateShareResp.
         let resp = leader_post!(
             self,
-            task_id,
+            &collect_req.task_id,
             task_config,
             "/aggregate_share",
             MEDIA_TYPE_AGG_SHARE_REQ,
@@ -361,7 +363,7 @@ pub trait DapLeader<S>: DapAuthorizedSender<S> + DapAggregator<S> {
         let collect_resp = CollectResp {
             encrypted_agg_shares: vec![leader_enc_agg_share, agg_share_resp.encrypted_agg_share],
         };
-        self.finish_collect_job(task_id, collect_id, &collect_resp)
+        self.finish_collect_job(&collect_req.task_id, collect_id, &collect_resp)
             .await?;
 
         // Mark reports as collected.
@@ -371,9 +373,9 @@ pub trait DapLeader<S>: DapAuthorizedSender<S> + DapAggregator<S> {
         Ok(agg_share_req.report_count)
     }
 
-    /// Fetch a set of reports grouped by task. For each task, first aggregate the reports, then
-    /// process the collect job queue. It is not safe to run multiple instances of this function in
-    /// parallel.
+    /// Fetch a set of reports grouped by task, then run an aggregation job for each task. once all
+    /// jobs completed, process the collect job queue. It is not safe to run multiple instances of
+    /// this function in parallel.
     ///
     /// This method is geared primarily towards testing. It also demonstrates how to properly
     /// synchronize collect and aggregation jobs. If used in a large DAP deployment, it is likely
@@ -385,31 +387,33 @@ pub trait DapLeader<S>: DapAuthorizedSender<S> + DapAggregator<S> {
     ) -> Result<DapLeaderProcessTelemetry, DapAbort> {
         let mut telem = DapLeaderProcessTelemetry::default();
 
+        // Fetch reports and run an aggregation job for each task.
+        //
         // TODO Handle tasks in parallel.
         for (task_id, reports) in self.get_reports(selector).await?.into_iter() {
             let task_config = self
                 .get_task_config_for(&task_id)
                 .ok_or(DapAbort::UnrecognizedTask)?;
 
-            // Aggregate the reports.
             telem.reports_processed += reports.len() as u64;
             if !reports.is_empty() {
                 telem.reports_aggregated +=
                     self.run_agg_job(&task_id, task_config, reports).await?;
             }
+        }
 
-            // Process pending collect jobs. We wait until all aggregation jobs are finished before
-            // proceeding to this step. This is to prevent a race condition involving an aggregate
-            // share computed during a collect job and any output shares computed during an aggregation
-            // job.
-            let pending = self.get_pending_collect_jobs(&task_id).await?;
-            for (collect_id, collect_reqs) in pending.into_iter() {
-                for collect_req in collect_reqs {
-                    telem.reports_collected += self
-                        .run_collect_job(&collect_id, &task_id, task_config, collect_req)
-                        .await?;
-                }
-            }
+        // Process pending collect jobs. We wait until all aggregation jobs are finished before
+        // proceeding to this step. This is to prevent a race condition involving an aggregate
+        // share computed during a collect job and any output shares computed during an aggregation
+        // job.
+        for (collect_id, collect_req) in self.get_pending_collect_jobs().await? {
+            let task_config = self
+                .get_task_config_for(&collect_req.task_id)
+                .ok_or(DapAbort::UnrecognizedTask)?;
+
+            telem.reports_collected += self
+                .run_collect_job(&collect_id, task_config, collect_req)
+                .await?;
         }
 
         Ok(telem)
diff --git a/daphne_worker/src/config.rs b/daphne_worker/src/config.rs
index 1f5953e..a9f6fec 100644
--- a/daphne_worker/src/config.rs
+++ b/daphne_worker/src/config.rs
@@ -9,7 +9,8 @@ use crate::{
     dap,
     durable::{
         aggregate_store::{durable_agg_store_name, DURABLE_AGGREGATE_STORE_DELETE_ALL},
-        leader_state_store::{durable_leader_state_name, DURABLE_LEADER_STATE_DELETE_ALL},
+        durable_queue_name,
+        leader_state_store::DURABLE_LEADER_STATE_DELETE_ALL,
         report_store::{durable_report_store_name, DURABLE_REPORT_STORE_DELETE_ALL},
     },
     int_err, now,
@@ -307,13 +308,9 @@ impl<D> DaphneWorkerConfig<D> {
 
         // Clear the leader's state.
         let namespace = self.durable_object("DAP_LEADER_STATE_STORE")?;
-        for task_id in self.tasks.keys() {
-            let stub = namespace
-                .id_from_name(&durable_leader_state_name(task_id))?
-                .get_stub()?;
-            // TODO Don't block on DO requests (issue multiple requests simultaneously).
-            durable_post!(stub, DURABLE_LEADER_STATE_DELETE_ALL, &()).await?;
-        }
+        let stub = namespace.id_from_name(&durable_queue_name(0))?.get_stub()?;
+        // TODO Don't block on DO requests (issue multiple requests simultaneously).
+        durable_post!(stub, DURABLE_LEADER_STATE_DELETE_ALL, &()).await?;
 
         Ok(())
     }
diff --git a/daphne_worker/src/dap.rs b/daphne_worker/src/dap.rs
index c1df0d6..ded58c3 100644
--- a/daphne_worker/src/dap.rs
+++ b/daphne_worker/src/dap.rs
@@ -12,21 +12,23 @@ use crate::{
         aggregate_store::{
             durable_agg_store_name, DURABLE_AGGREGATE_STORE_GET, DURABLE_AGGREGATE_STORE_MERGE,
         },
+        durable_queue_name,
         helper_state_store::{
             durable_helper_state_name, DURABLE_HELPER_STATE_GET, DURABLE_HELPER_STATE_PUT,
         },
+        leader_agg_job_queue::DURABLE_LEADER_AGG_JOB_QUEUE_GET_PENDING,
         leader_state_store::{
-            durable_leader_state_name, LeaderStateStoreUpdateCollectReq,
-            DURABLE_LEADER_STATE_FINISH_COLLECT_REQ, DURABLE_LEADER_STATE_GET_COLLECT_REQS,
-            DURABLE_LEADER_STATE_GET_COLLECT_RESP, DURABLE_LEADER_STATE_PUT_COLLECT_REQ,
+            LeaderStateStoreUpdateCollectReq, DURABLE_LEADER_STATE_FINISH_COLLECT_REQ,
+            DURABLE_LEADER_STATE_GET_COLLECT_REQS, DURABLE_LEADER_STATE_GET_COLLECT_RESP,
+            DURABLE_LEADER_STATE_PUT_COLLECT_REQ,
         },
         report_store::{
-            ReportStoreGetPending, ReportStoreResult, DURABLE_REPORT_STORE_GET_PENDING,
+            ReportStoreResult, DURABLE_REPORT_STORE_GET_PENDING,
             DURABLE_REPORT_STORE_MARK_COLLECTED, DURABLE_REPORT_STORE_PUT_PENDING,
             DURABLE_REPORT_STORE_PUT_PROCESSED,
         },
     },
-    now, InternalAggregateInfo,
+    InternalAggregateInfo,
 };
 use async_trait::async_trait;
 use daphne::{
@@ -249,40 +251,32 @@ impl<D> DapLeader<BearerToken> for DaphneWorkerConfig<D> {
         &self,
         selector: &InternalAggregateInfo,
     ) -> std::result::Result<HashMap<Id, Vec<Report>>, DapError> {
-        let durable_namespace = self.durable_object("DAP_REPORT_STORE")?;
-
-        let task_config = self
-            .get_task_config_for(&selector.task_id)
-            .ok_or_else(|| DapError::fatal(INT_ERR_UNRECOGNIZED_TASK))?;
-
-        let batch_interval = if let Some(ref interval) = selector.batch_info {
-            if !interval.is_valid_for(task_config) {
-                return Err(DapError::fatal(INT_ERR_INVALID_BATCH_INTERVAL));
-            }
-            interval.clone()
-        } else {
-            task_config.current_batch_window(now())
-        };
-
-        // Fetch a candidate report set.
+        // Read at most `selector.max_buckets` buckets from the agg job queue. The result is ordered
+        // from oldest to newest.
         //
-        // TODO Start this loop at the previously visited bucket.
+        // NOTE There is only one agg job queue for now (`queue_num == 0`). In the future, work
+        // will be sharded across multiple queues.
+        let namespace = self.durable_object("DAP_LEADER_AGG_JOB_QUEUE")?;
+        let stub = namespace.id_from_name(&durable_queue_name(0))?.get_stub()?;
+        let mut resp = durable_post!(
+            stub,
+            DURABLE_LEADER_AGG_JOB_QUEUE_GET_PENDING,
+            &selector.max_buckets
+        )
+        .await?;
+        let res: Vec<String> = resp.json().await?;
+
+        // Drain at most `selector.max_reports` from each bucket.
         //
-        // TODO We can save latency here if the input shares were already decrypted. This would have
-        // the added benefit of reducing data loss caused by HPKE key rotation. (It would be nice if
-        // the Helper could preprocess its input shares in the same way. See
-        // https://github.com/abetterinternet/ppm-specification/pull/174.)
-        let mut reports = Vec::with_capacity(selector.agg_rate.try_into().unwrap());
-        for durable_name in self.iter_report_store_names(&selector.task_id, &batch_interval)? {
-            let num_reports_remaining = selector.agg_rate - reports.len() as u64;
-            let stub = durable_namespace.id_from_name(&durable_name)?.get_stub()?;
-            // TODO Don't block on DO request (issue multiple requests simultaneously).
+        // TODO Figure out if we can safely handle each bucket in parallel.
+        let namespace = self.durable_object("DAP_REPORT_STORE")?;
+        let mut reports_per_task: HashMap<Id, Vec<Report>> = HashMap::new();
+        for report_store_id_hex in res.into_iter() {
+            let stub = namespace.id_from_string(&report_store_id_hex)?.get_stub()?;
             let reports_from_durable: Vec<String> = durable_post!(
                 stub,
                 DURABLE_REPORT_STORE_GET_PENDING,
-                &ReportStoreGetPending {
-                    reports_requested: num_reports_remaining,
-                }
+                &selector.max_reports
             )
             .await?
             .json()
@@ -293,20 +287,16 @@ impl<D> DapLeader<BearerToken> for DaphneWorkerConfig<D> {
                     Report::get_decoded(&hex::decode(&report_hex).map_err(|_| {
                         DapError::fatal("response from ReportStore is not valid hex")
                     })?)?;
-                reports.push(report);
-            }
 
-            if reports.len() as u64 > selector.agg_rate {
-                return Err(DapError::fatal(
-                    "number of reports received from report store exceeds the number requested",
-                ));
-            }
-            if reports.len() as u64 == selector.agg_rate {
-                break;
+                if let Some(reports) = reports_per_task.get_mut(&report.task_id) {
+                    reports.push(report);
+                } else {
+                    reports_per_task.insert(report.task_id.clone(), vec![report]);
+                }
             }
         }
 
-        Ok(HashMap::from([(selector.task_id.clone(), reports)]))
+        Ok(reports_per_task)
     }
 
     async fn init_collect_job(
@@ -318,9 +308,7 @@ impl<D> DapLeader<BearerToken> for DaphneWorkerConfig<D> {
             .ok_or_else(|| DapError::fatal(INT_ERR_UNRECOGNIZED_TASK))?;
 
         let namespace = self.durable_object("DAP_LEADER_STATE_STORE")?;
-        let stub = namespace
-            .id_from_name(&durable_leader_state_name(&collect_req.task_id))?
-            .get_stub()?;
+        let stub = namespace.id_from_name(&durable_queue_name(0))?.get_stub()?;
         let collect_id: Id =
             durable_post!(stub, DURABLE_LEADER_STATE_PUT_COLLECT_REQ, &collect_req)
                 .await?
@@ -341,13 +329,11 @@ impl<D> DapLeader<BearerToken> for DaphneWorkerConfig<D> {
 
     async fn poll_collect_job(
         &self,
-        task_id: &Id,
+        _task_id: &Id,
         collect_id: &Id,
     ) -> std::result::Result<DapCollectJob, DapError> {
         let namespace = self.durable_object("DAP_LEADER_STATE_STORE")?;
-        let stub = namespace
-            .id_from_name(&durable_leader_state_name(task_id))?
-            .get_stub()?;
+        let stub = namespace.id_from_name(&durable_queue_name(0))?.get_stub()?;
         let res: DapCollectJob =
             durable_post!(stub, DURABLE_LEADER_STATE_GET_COLLECT_RESP, &collect_id)
                 .await?
@@ -358,13 +344,12 @@ impl<D> DapLeader<BearerToken> for DaphneWorkerConfig<D> {
 
     async fn get_pending_collect_jobs(
         &self,
-        task_id: &Id,
-    ) -> std::result::Result<HashMap<Id, Vec<CollectReq>>, DapError> {
+    ) -> std::result::Result<Vec<(Id, CollectReq)>, DapError> {
         let leader_state_namespace = self.durable_object("DAP_LEADER_STATE_STORE")?;
         let leader_state_stub = leader_state_namespace
-            .id_from_name(&durable_leader_state_name(task_id))?
+            .id_from_name(&durable_queue_name(0))?
             .get_stub()?;
-        let res: HashMap<Id, Vec<CollectReq>> =
+        let res: Vec<(Id, CollectReq)> =
             durable_get!(leader_state_stub, DURABLE_LEADER_STATE_GET_COLLECT_REQS)
                 .await?
                 .json()
@@ -374,13 +359,13 @@ impl<D> DapLeader<BearerToken> for DaphneWorkerConfig<D> {
 
     async fn finish_collect_job(
         &self,
-        task_id: &Id,
+        _task_id: &Id,
         collect_id: &Id,
         collect_resp: &CollectResp,
     ) -> std::result::Result<(), DapError> {
         let leader_state_namespace = self.durable_object("DAP_LEADER_STATE_STORE")?;
         let leader_state_stub = leader_state_namespace
-            .id_from_name(&durable_leader_state_name(task_id))?
+            .id_from_name(&durable_queue_name(0))?
             .get_stub()?;
         durable_post!(
             leader_state_stub,
diff --git a/daphne_worker/src/durable/leader_agg_job_queue.rs b/daphne_worker/src/durable/leader_agg_job_queue.rs
new file mode 100644
index 0000000..2d1952b
--- /dev/null
+++ b/daphne_worker/src/durable/leader_agg_job_queue.rs
@@ -0,0 +1,103 @@
+// Copyright (c) 2022 Cloudflare, Inc. All rights reserved.
+// SPDX-License-Identifier: BSD-3-Clause
+
+use crate::int_err;
+use serde::{Deserialize, Serialize};
+use worker::*;
+
+pub(crate) const DURABLE_LEADER_AGG_JOB_QUEUE_DELETE_ALL: &str =
+    "/internal/do/agg_job_queue/delete_all";
+pub(crate) const DURABLE_LEADER_AGG_JOB_QUEUE_PUT_PENDING: &str =
+    "/internal/do/agg_job_queue/put_pending";
+pub(crate) const DURABLE_LEADER_AGG_JOB_QUEUE_GET_PENDING: &str =
+    "/internal/do/agg_job_queue/get_pending";
+pub(crate) const DURABLE_LEADER_AGG_JOB_QUEUE_FINISH: &str = "/internal/do/agg_job_queue/finish";
+
+/// An aggregation job.
+#[derive(Serialize, Deserialize)]
+pub(crate) struct AggregationJob {
+    /// Durable object ID of the [`ReportStore`](crate::durable::ReportStore) instance with reports
+    /// to be aggregated.
+    pub(crate) report_store_id_hex: String,
+
+    /// The time at which the aggregation job was scheduled. Aggregation jobs are processed in
+    /// order of arrival. [`ReportStore`](crate::durable::ReportStore) sets this to the time the
+    /// first report arrived in the bucket.
+    pub(crate) time: u64,
+
+    /// A random value used to break ties between jobs with the same time stamp.
+    pub(crate) rand: u64,
+}
+
+/// Durable Object (DO) representing an aggregation job queue.
+///
+/// An instance of the [`LeaderAggregationJobQueue`] DO is named `/queue/<queue_num>` where
+/// `<queue_num>` is an integer idnentifying a specific queue.
+#[durable_object]
+pub struct LeaderAggregationJobQueue {
+    #[allow(dead_code)]
+    state: State,
+}
+
+#[durable_object]
+impl DurableObject for LeaderAggregationJobQueue {
+    fn new(state: State, _env: Env) -> Self {
+        Self { state }
+    }
+
+    async fn fetch(&mut self, mut req: Request) -> Result<Response> {
+        match (req.path().as_ref(), req.method()) {
+            (DURABLE_LEADER_AGG_JOB_QUEUE_DELETE_ALL, Method::Post) => {
+                self.state.storage().delete_all().await?;
+                Response::empty()
+            }
+
+            (DURABLE_LEADER_AGG_JOB_QUEUE_PUT_PENDING, Method::Post) => {
+                let agg_job: AggregationJob = req.json().await?;
+                self.state
+                    .storage()
+                    .put(&bucket_key_for(&agg_job), agg_job.report_store_id_hex)
+                    .await?;
+                Response::empty()
+            }
+
+            (DURABLE_LEADER_AGG_JOB_QUEUE_GET_PENDING, Method::Post) => {
+                let num_buckets: usize = req.json().await?;
+                let opt = ListOptions::new().prefix("bucket/").limit(num_buckets);
+                let iter = self.state.storage().list_with_options(opt).await?.entries();
+                let mut item = iter.next()?;
+                let mut res = Vec::new();
+                while !item.done() {
+                    let (_bucket_key, report_store_id_hex): (String, String) =
+                        item.value().into_serde()?;
+                    res.push(report_store_id_hex);
+                    item = iter.next()?;
+                }
+
+                // Results are in order of the oldest bucket first.
+                Response::from_json(&res)
+            }
+
+            (DURABLE_LEADER_AGG_JOB_QUEUE_FINISH, Method::Post) => {
+                let agg_job: AggregationJob = req.json().await?;
+                self.state
+                    .storage()
+                    .delete(&bucket_key_for(&agg_job))
+                    .await?;
+                Response::empty()
+            }
+
+            _ => Err(int_err(format!(
+                "LeaderAggregationJobQueue: unexpected request: method={:?}; path={:?}",
+                req.method(),
+                req.path()
+            ))),
+        }
+    }
+}
+
+fn bucket_key_for(agg_job: &AggregationJob) -> String {
+    // Format the bucket so that they are processed in order of arrival (oldest first). The
+    // timestamp is used for this purpose; the randomm value is used for breaking ties.
+    format!("bucket/time/{:020}/rand/{:020}", agg_job.time, agg_job.rand)
+}
diff --git a/daphne_worker/src/durable/leader_state_store.rs b/daphne_worker/src/durable/leader_state_store.rs
index ff1c6ed..bff9293 100644
--- a/daphne_worker/src/durable/leader_state_store.rs
+++ b/daphne_worker/src/durable/leader_state_store.rs
@@ -14,16 +14,9 @@ use prio::{
     vdaf::prg::{Prg, PrgAes128, Seed, SeedStream},
 };
 use serde::{Deserialize, Serialize};
-use std::{
-    collections::{HashMap, VecDeque},
-    convert::TryInto,
-};
+use std::{collections::VecDeque, convert::TryInto};
 use worker::*;
 
-pub(crate) fn durable_leader_state_name(task_id: &Id) -> String {
-    format!("/task/{}", task_id.to_base64url())
-}
-
 pub(crate) const DURABLE_LEADER_STATE_DELETE_ALL: &str = "/internal/do/leader_state/delete_all";
 pub(crate) const DURABLE_LEADER_STATE_PUT_COLLECT_REQ: &str =
     "/internal/do/leader_state/put_collect_req";
@@ -48,8 +41,8 @@ struct OrderedCollectReq {
 
 /// Durable Object (DO) for storing the Leader's state for a given task.
 ///
-/// An instance of the [`LeaderStateStore`] DO is named `/task/<task_id>`, where `<task_id>` is
-/// the task ID.
+/// An instance of the [`LeaderStateStore`] DO is named `/queue/<queue_num>`, where `<queue_num>`
+/// is an integer representing a specific queue.
 //
 // TODO spec: Consider allowing completed aggregate results to be deleted after a period of time.
 #[durable_object]
@@ -148,17 +141,11 @@ impl DurableObject for LeaderStateStore {
                 }
 
                 list.sort_unstable_by_key(|(_id, prioritized)| prioritized.priority);
-                let mut list: VecDeque<(Id, CollectReq)> = list
+                let list: VecDeque<(Id, CollectReq)> = list
                     .into_iter()
                     .map(|(id, ordered_collect_req)| (id, ordered_collect_req.collect_req))
                     .collect();
-                let mut res: HashMap<Id, Vec<CollectReq>> = HashMap::default();
-                while !list.is_empty() {
-                    let (id, collect_req) = list.pop_front().unwrap();
-                    let collect_reqs = res.entry(id).or_insert_with(Vec::default);
-                    collect_reqs.push(collect_req);
-                }
-                Response::from_json(&res)
+                Response::from_json(&list)
             }
 
             // Store a CollectResp corresponding to a pending CollectReq.
diff --git a/daphne_worker/src/durable/mod.rs b/daphne_worker/src/durable/mod.rs
index d32ab92..62da78b 100644
--- a/daphne_worker/src/durable/mod.rs
+++ b/daphne_worker/src/durable/mod.rs
@@ -32,8 +32,13 @@ pub(crate) async fn state_get<T: for<'a> Deserialize<'a>>(
     })
 }
 
+pub(crate) fn durable_queue_name(queue_num: usize) -> String {
+    format!("/queue/{}", queue_num)
+}
+
 pub(crate) mod aggregate_store;
 pub(crate) mod helper_state_store;
+pub(crate) mod leader_agg_job_queue;
 pub(crate) mod leader_state_store;
 pub(crate) mod report_store;
 #[cfg(test)]
diff --git a/daphne_worker/src/durable/report_store.rs b/daphne_worker/src/durable/report_store.rs
index ea910b5..b56c1c7 100644
--- a/daphne_worker/src/durable/report_store.rs
+++ b/daphne_worker/src/durable/report_store.rs
@@ -1,10 +1,20 @@
 // Copyright (c) 2022 Cloudflare, Inc. All rights reserved.
 // SPDX-License-Identifier: BSD-3-Clause
 
-use crate::{durable::state_get_or_default, int_err};
+use crate::{
+    durable::{
+        durable_queue_name,
+        leader_agg_job_queue::{
+            AggregationJob, DURABLE_LEADER_AGG_JOB_QUEUE_FINISH,
+            DURABLE_LEADER_AGG_JOB_QUEUE_PUT_PENDING,
+        },
+        state_get, state_get_or_default,
+    },
+    int_err, now,
+};
 use daphne::messages::TransitionFailure;
+use rand::prelude::*;
 use serde::{Deserialize, Serialize};
-use std::convert::TryInto;
 use worker::*;
 
 pub(crate) fn durable_report_store_name(
@@ -26,11 +36,6 @@ pub(crate) const DURABLE_REPORT_STORE_PUT_PROCESSED: &str =
 pub(crate) const DURABLE_REPORT_STORE_MARK_COLLECTED: &str =
     "/internal/do/report_store/mark_collected";
 
-#[derive(Deserialize, Serialize)]
-pub(crate) struct ReportStoreGetPending {
-    pub(crate) reports_requested: u64,
-}
-
 #[derive(Deserialize, Serialize)]
 pub(crate) enum ReportStoreResult {
     Ok,
@@ -51,6 +56,7 @@ pub(crate) enum ReportStoreResult {
 pub struct ReportStore {
     #[allow(dead_code)]
     state: State,
+    env: Env,
 }
 
 impl ReportStore {
@@ -70,11 +76,12 @@ impl ReportStore {
 
 #[durable_object]
 impl DurableObject for ReportStore {
-    fn new(state: State, _env: Env) -> Self {
-        Self { state }
+    fn new(state: State, env: Env) -> Self {
+        Self { state, env }
     }
 
     async fn fetch(&mut self, mut req: Request) -> Result<Response> {
+        let mut rng = thread_rng();
         match (req.path().as_ref(), req.method()) {
             (DURABLE_REPORT_STORE_DELETE_ALL, Method::Post) => {
                 self.state.storage().delete_all().await?;
@@ -82,8 +89,7 @@ impl DurableObject for ReportStore {
             }
 
             (DURABLE_REPORT_STORE_GET_PENDING, Method::Post) => {
-                let info: ReportStoreGetPending = req.json().await?;
-                let reports_requested = info.reports_requested.try_into().unwrap();
+                let reports_requested: usize = req.json().await?;
                 let opt = ListOptions::new()
                     .prefix("pending/")
                     .limit(reports_requested);
@@ -98,9 +104,22 @@ impl DurableObject for ReportStore {
                     item = iter.next()?;
                 }
 
+                // If this bucket is empty, then remove it from the agg job queue.
+                if reports.is_empty() {
+                    let agg_job: Option<AggregationJob> = state_get(&self.state, "agg_job").await?;
+                    if let Some(agg_job) = agg_job {
+                        // NOTE There is only one agg job queue for now. In the future, work will
+                        // be sharded across multiple queues.
+                        let namespace = self.env.durable_object("DAP_LEADER_AGG_JOB_QUEUE")?;
+                        let stub = namespace.id_from_name(&durable_queue_name(0))?.get_stub()?;
+                        durable_post!(stub, DURABLE_LEADER_AGG_JOB_QUEUE_FINISH, &agg_job).await?;
+                        self.state.storage().delete("agg_job").await?;
+                    }
+                }
+
                 // NOTE In order to support DAP tasks that require longer batch lifetimes, it will
                 // necessary to check if the lifetime has been reached before removing reports from
-                // storage.
+                // storage. We might consider putting reports in KV instead.
                 self.state.storage().delete_multiple(keys).await?;
                 Response::from_json(&reports)
             }
@@ -115,6 +134,25 @@ impl DurableObject for ReportStore {
                     let key = format!("pending/{}", nonce_hex);
                     self.state.storage().put(&key, report_hex).await?;
                 }
+
+                // Check if processing for this bucket of reports has been scheduled. If so, add
+                // this bucket to the aggregation job queue.
+                let agg_job: Option<AggregationJob> = state_get(&self.state, "agg_job").await?;
+                if agg_job.is_none() {
+                    let agg_job = AggregationJob {
+                        report_store_id_hex: self.state.id().to_string(),
+                        time: now(),
+                        rand: rng.gen(),
+                    };
+
+                    let namespace = self.env.durable_object("DAP_LEADER_AGG_JOB_QUEUE")?;
+                    // TODO Shard the work across multiple job queues rather than just one. (See
+                    // issue #25.) For now there is jsut one job queue.
+                    let stub = namespace.id_from_name(&durable_queue_name(0))?.get_stub()?;
+                    durable_post!(stub, DURABLE_LEADER_AGG_JOB_QUEUE_PUT_PENDING, &agg_job).await?;
+                    self.state.storage().put("agg_job", agg_job).await?;
+                }
+
                 Response::from_json(&res)
             }
 
diff --git a/daphne_worker/src/lib.rs b/daphne_worker/src/lib.rs
index 299ab7c..8ec436d 100644
--- a/daphne_worker/src/lib.rs
+++ b/daphne_worker/src/lib.rs
@@ -20,13 +20,11 @@ use worker::*;
 /// Parameters used by the Leader to select a set of reports for aggregation.
 #[derive(Debug, Deserialize, Serialize)]
 pub struct InternalAggregateInfo {
-    pub task_id: Id,
+    /// Maximum number of buckets to process at once.
+    pub max_buckets: u64,
 
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub batch_info: Option<Interval>,
-
-    /// The number of reports to aggregate in one execution of the aggregation flow.
-    pub agg_rate: u64,
+    /// Maximum number of reports to drain from each bucket.
+    pub max_reports: u64,
 }
 
 macro_rules! durable_get {
@@ -150,14 +148,6 @@ pub async fn run_daphne_worker(req: Request, env: Env) -> Result<Response> {
                 .get_async(
                     "/collect/task/:task_id/req/:collect_id",
                     |_req, ctx| async move {
-                        // TODO(cjpatton) Decide if this request needs to be authorized using the
-                        // Collector's bearer token.
-                        //
-                        // Right now the spec does not require it. (It might not even allow it!)
-                        // This would be reasonable if this request doesn't change the Aggregator's
-                        // state, however in our case it does: The durable object deletes the
-                        // aggregate shares after they've been collected. Thus we'll have to either
-                        // modify this endpoint so that it is stateless or add authorization.
                         let task_id = parse_id!(ctx.param("task_id"));
                         let collect_id = parse_id!(ctx.param("collect_id"));
                         let config = DaphneWorkerConfig::from_worker_context(ctx)?;


Test patch:
diff --git a/daphne/src/roles_test.rs b/daphne/src/roles_test.rs
index d8781bf..94bf694 100644
--- a/daphne/src/roles_test.rs
+++ b/daphne/src/roles_test.rs
@@ -534,8 +534,8 @@ async fn poll_collect_job_test_results() {
     );
 
     // Leader: Get pending collect job to obtain collect_id
-    let resp = leader.get_pending_collect_jobs(&task_id).await.unwrap();
-    let collect_id = resp.keys().collect::<Vec<&Id>>()[0];
+    let resp = leader.get_pending_collect_jobs().await.unwrap();
+    let (collect_id, _collect_req) = &resp[0];
     let collect_resp = CollectResp {
         encrypted_agg_shares: Vec::default(),
     };
@@ -585,12 +585,11 @@ async fn http_post_collect_success() {
 
     // Leader: Handle the CollectReq received from Collector.
     let url = leader.http_post_collect(&req).await.unwrap();
-    let mut resp = leader.get_pending_collect_jobs(&task_id).await.unwrap();
-    let leader_collect_id = resp.keys().collect::<Vec<&Id>>()[0].clone();
-    let leader_collect_req = resp.remove(&leader_collect_id).unwrap();
+    let resp = leader.get_pending_collect_jobs().await.unwrap();
+    let (leader_collect_id, leader_collect_req) = &resp[0];
 
     // Check that the CollectReq sent by Collector is the same that is received by Leader.
-    assert_eq!(collector_collect_req, leader_collect_req[0]);
+    assert_eq!(&collector_collect_req, leader_collect_req);
 
     // Check that the collect_id included in the URI is the same with the one received
     // by Leader.
@@ -714,9 +713,8 @@ async fn e2e() {
 
     // Leader: Handle the CollectReq received from Collector.
     leader.http_post_collect(&req).await.unwrap();
-    let mut resp = leader.get_pending_collect_jobs(&task_id).await.unwrap();
-    let collect_id = resp.keys().collect::<Vec<&Id>>()[0].clone();
-    let collect_req = resp.remove(&collect_id).unwrap()[0].clone();
+    let resp = leader.get_pending_collect_jobs().await.unwrap();
+    let (collect_id, collect_req) = &resp[0];
 
     // Leader: Get Leader's encrypted aggregate share.
     let leader_agg_share = leader
@@ -736,9 +734,9 @@ async fn e2e() {
 
     // Leader: Prepare AggregateShareReq.
     let agg_share_req = AggregateShareReq {
-        task_id: collect_req.task_id,
-        batch_interval: collect_req.batch_interval,
-        agg_param: collect_req.agg_param,
+        task_id: collect_req.task_id.clone(),
+        batch_interval: collect_req.batch_interval.clone(),
+        agg_param: collect_req.agg_param.clone(),
         report_count: leader_agg_share.report_count,
         checksum: leader_agg_share.checksum,
     };
diff --git a/daphne/src/testing.rs b/daphne/src/testing.rs
index d8053ba..4f4d6cf 100644
--- a/daphne/src/testing.rs
+++ b/daphne/src/testing.rs
@@ -537,29 +537,22 @@ impl DapLeader<BearerToken> for MockAggregator {
     }
 
     // Called to retrieve pending CollectReq.
-    async fn get_pending_collect_jobs(
-        &self,
-        task_id: &Id,
-    ) -> Result<HashMap<Id, Vec<CollectReq>>, DapError> {
+    async fn get_pending_collect_jobs(&self) -> Result<Vec<(Id, CollectReq)>, DapError> {
         let mut leader_state_store_mutex_guard = self
             .leader_state_store
             .lock()
             .map_err(|e| DapError::Fatal(e.to_string()))?;
         let leader_state_store = leader_state_store_mutex_guard.deref_mut();
 
-        let leader_state = leader_state_store
-            .get_mut(task_id)
-            .ok_or_else(|| DapError::fatal("collect job not found for task_id"))?;
-        let mut res: HashMap<Id, Vec<CollectReq>> = HashMap::default();
-
-        // Iterate over collect IDs and copy them and their associated requests to the response.
-        for collect_id in leader_state.collect_ids.iter() {
-            if let CollectJobState::Pending(collect_req) =
-                leader_state.collect_jobs.get(collect_id).unwrap()
-            {
-                res.entry(collect_id.clone())
-                    .or_insert_with(Vec::default)
-                    .push(collect_req.clone());
+        let mut res = Vec::new();
+        for (_task_id, leader_state) in leader_state_store.iter() {
+            // Iterate over collect IDs and copy them and their associated requests to the response.
+            for collect_id in leader_state.collect_ids.iter() {
+                if let CollectJobState::Pending(collect_req) =
+                    leader_state.collect_jobs.get(collect_id).unwrap()
+                {
+                    res.push((collect_id.clone(), collect_req.clone()));
+                }
             }
         }
         Ok(res)
diff --git a/daphne_worker_test/tests/e2e.rs b/daphne_worker_test/tests/e2e.rs
index aec1363..8c15885 100644
--- a/daphne_worker_test/tests/e2e.rs
+++ b/daphne_worker_test/tests/e2e.rs
@@ -14,7 +14,6 @@ use daphne::{
 use daphne_worker::InternalAggregateInfo;
 use prio::codec::{Decode, Encode};
 use rand::prelude::*;
-use std::time::SystemTime;
 use test_runner::{TestRunner, COLLECTOR_BEARER_TOKEN, COLLECTOR_HPKE_RECEIVER_CONFIG};
 
 #[tokio::test]
@@ -174,16 +173,15 @@ async fn e2e_internal_leader_process() {
     let hpke_config_list = t.get_hpke_configs(&client).await;
 
     let agg_info = InternalAggregateInfo {
-        task_id: t.task_id.clone(),
-        batch_info: t.batch_info(),
-        agg_rate: t.min_batch_size,
+        max_buckets: 100, // Needs to be sufficiently large to touch each bucket.
+        max_reports: t.min_batch_size,
     };
 
-    let batch_interval = agg_info.batch_info.as_ref().unwrap();
+    let batch_interval = t.batch_interval();
 
     // Upload a number of reports (a few more than the aggregation rate).
     let mut rng = thread_rng();
-    for _ in 0..agg_info.agg_rate + 3 {
+    for _ in 0..agg_info.max_reports + 3 {
         let now = rng.gen_range(batch_interval.start..batch_interval.end());
         t.leader_post_expect_ok(
             &client,
@@ -197,67 +195,31 @@ async fn e2e_internal_leader_process() {
         .await;
     }
 
-    // Run an iteration of the processing loop. Expect the number of aggregated reports to match
-    // the aggregation rate.
-    let agg_telem = t.internal_process(&client, &agg_info).await;
-    assert_eq!(agg_telem.reports_aggregated, agg_info.agg_rate);
-
-    // Run a second iteration of the processing loop. Expect the number of aggregated reports to
-    // be less than the aggregation rate.
     let agg_telem = t.internal_process(&client, &agg_info).await;
-    assert_eq!(agg_telem.reports_aggregated, 3);
-
-    // Run a third iteration of the processing loop. By now the report store has been drained, so
-    // no reports should have been aggregated.
-    let agg_telem = t.internal_process(&client, &agg_info).await;
-    assert_eq!(agg_telem.reports_aggregated, 0);
-}
-
-// Test leader processing loop for the current batch window.
-//
-// WARNING: flaky test: This test assumes the client and aggregators have synchronized clocks. It
-// may fail if the current time is close to the end of the current batch window, which is
-// determined by the minimum batch interval.
-#[tokio::test]
-#[cfg_attr(not(feature = "test_e2e"), ignore)]
-async fn e2e_internal_leader_process_current_batch_window() {
-    let t = TestRunner::default().await;
-    let client = t.http_client();
-    let hpke_config_list = t.get_hpke_configs(&client).await;
-
-    let agg_info = InternalAggregateInfo {
-        task_id: t.task_id.clone(),
-        batch_info: None,
-        agg_rate: 1,
-    };
-
-    // Upload a report using the current time.
-    let now = SystemTime::now()
-        .duration_since(SystemTime::UNIX_EPOCH)
-        .unwrap()
-        .as_secs();
-    t.leader_post_expect_ok(
-        &client,
-        "/upload",
-        constants::MEDIA_TYPE_REPORT,
-        t.vdaf
-            .produce_report(&hpke_config_list, now, &t.task_id, DapMeasurement::U64(1))
-            .unwrap()
-            .get_encoded(),
-    )
-    .await;
+    assert_eq!(
+        agg_telem.reports_processed,
+        agg_info.max_reports + 3,
+        "reports processed"
+    );
+    assert_eq!(
+        agg_telem.reports_aggregated,
+        agg_info.max_reports + 3,
+        "reports aggregated"
+    );
+    assert_eq!(agg_telem.reports_collected, 0, "reports collected");
 
+    // There should be nothing left to aggregate.
     let agg_telem = t.internal_process(&client, &agg_info).await;
-    t.internal_reset(&None).await;
-    assert_eq!(agg_telem.reports_aggregated, 1);
+    assert_eq!(agg_telem.reports_processed, 0, "reports processed");
+    assert_eq!(agg_telem.reports_aggregated, 0, "reports aggregated");
+    assert_eq!(agg_telem.reports_collected, 0, "reports collected");
 }
 
 #[tokio::test]
 #[cfg_attr(not(feature = "test_e2e"), ignore)]
 async fn e2e_leader_collect_ok() {
     let t = TestRunner::default().await;
-    let batch_info = t.batch_info();
-    let batch_interval = batch_info.as_ref().unwrap();
+    let batch_interval = t.batch_interval();
 
     let client = t.http_client();
     let hpke_config_list = t.get_hpke_configs(&client).await;
@@ -298,15 +260,23 @@ async fn e2e_leader_collect_ok() {
         .internal_process(
             &client,
             &InternalAggregateInfo {
-                task_id: t.task_id.clone(),
-                batch_info: batch_info.clone(),
-                agg_rate: 100,
+                max_buckets: 100, // Needs to be sufficiently large to touch each bucket.
+                max_reports: 100,
             },
         )
         .await;
-    assert_eq!(agg_telem.reports_processed, t.min_batch_size);
-    assert_eq!(agg_telem.reports_aggregated, t.min_batch_size);
-    assert_eq!(agg_telem.reports_collected, t.min_batch_size);
+    assert_eq!(
+        agg_telem.reports_processed, t.min_batch_size,
+        "reports processed"
+    );
+    assert_eq!(
+        agg_telem.reports_aggregated, t.min_batch_size,
+        "reports aggregated"
+    );
+    assert_eq!(
+        agg_telem.reports_collected, t.min_batch_size,
+        "reports collected"
+    );
 
     // Poll the collect URI.
     let resp = client.get(collect_uri.as_str()).send().await.unwrap();
@@ -349,13 +319,67 @@ async fn e2e_leader_collect_ok() {
     .await;
 }
 
+// Test that collect jobs complete even if the request is issued after all reports for the task
+// have been processed.
 #[tokio::test]
 #[cfg_attr(not(feature = "test_e2e"), ignore)]
-async fn e2e_leader_collect_not_ready_min_batch_size() {
+async fn e2e_leader_collect_ok_interleaved() {
     let t = TestRunner::default().await;
-    let batch_info = t.batch_info();
-    let batch_interval = batch_info.as_ref().unwrap();
+    let client = t.http_client();
+    let batch_interval = t.batch_interval();
+    let hpke_config_list = t.get_hpke_configs(&client).await;
 
+    // The reports are uploaded in the background.
+    let mut rng = thread_rng();
+    for _ in 0..t.min_batch_size {
+        let now = rng.gen_range(batch_interval.start..batch_interval.end());
+        t.leader_post_expect_ok(
+            &client,
+            "/upload",
+            constants::MEDIA_TYPE_REPORT,
+            t.vdaf
+                .produce_report(&hpke_config_list, now, &t.task_id, DapMeasurement::U64(1))
+                .unwrap()
+                .get_encoded(),
+        )
+        .await;
+    }
+
+    let agg_info = InternalAggregateInfo {
+        max_buckets: 100, // Needs to be sufficiently large to touch each bucket.
+        max_reports: 100,
+    };
+
+    // All reports for the task get processed ...
+    let agg_telem = t.internal_process(&client, &agg_info).await;
+    assert_eq!(
+        agg_telem.reports_processed, t.min_batch_size,
+        "reports processed"
+    );
+
+    // ... then the collect request is issued ...
+    let collect_req = CollectReq {
+        task_id: t.task_id.clone(),
+        batch_interval: batch_interval.clone(),
+        agg_param: Vec::new(),
+    };
+    let _collect_uri = t
+        .leader_post_collect(&client, collect_req.get_encoded())
+        .await;
+
+    // ... then the collect job gets completed.
+    let agg_telem = t.internal_process(&client, &agg_info).await;
+    assert_eq!(
+        agg_telem.reports_collected, t.min_batch_size,
+        "reports collected"
+    );
+}
+
+#[tokio::test]
+#[cfg_attr(not(feature = "test_e2e"), ignore)]
+async fn e2e_leader_collect_not_ready_min_batch_size() {
+    let t = TestRunner::default().await;
+    let batch_interval = t.batch_interval();
     let client = t.http_client();
     let hpke_config_list = t.get_hpke_configs(&client).await;
 
@@ -391,9 +415,8 @@ async fn e2e_leader_collect_not_ready_min_batch_size() {
         .internal_process(
             &client,
             &InternalAggregateInfo {
-                task_id: t.task_id.clone(),
-                batch_info,
-                agg_rate: 100,
+                max_buckets: 100, // Needs to be sufficiently large to touch each bucket.
+                max_reports: 100,
             },
         )
         .await;
@@ -431,7 +454,7 @@ async fn e2e_leader_collect_abort_unknown_request() {
 async fn e2e_leader_collect_abort_invalid_batch_interval() {
     let t = TestRunner::default().await;
     let client = t.http_client();
-    let batch_interval = t.batch_info().unwrap();
+    let batch_interval = t.batch_interval();
 
     // Start of batch interval does not align with min_batch_duration.
     let collect_req = CollectReq {
diff --git a/daphne_worker_test/tests/janus.rs b/daphne_worker_test/tests/janus.rs
index 579ce7a..913b177 100644
--- a/daphne_worker_test/tests/janus.rs
+++ b/daphne_worker_test/tests/janus.rs
@@ -60,9 +60,8 @@ async fn janus_client() {
     janus_client.upload(&23).await.unwrap();
 
     let agg_info = InternalAggregateInfo {
-        task_id: t.task_id.clone(),
-        batch_info: t.batch_info(),
-        agg_rate: 1,
+        max_buckets: 100, // Needs to be large enough to touch each bucket.
+        max_reports: 1,
     };
     let agg_telem = t.internal_process(&client, &agg_info).await;
     assert_eq!(agg_telem.reports_processed, 1);
@@ -80,15 +79,14 @@ async fn janus_helper() {
     let client = t.http_client();
     let hpke_config_list = t.get_hpke_configs(&client).await;
     let agg_info = InternalAggregateInfo {
-        task_id: t.task_id.clone(),
-        batch_info: t.batch_info(),
-        agg_rate: t.min_batch_size,
+        max_buckets: 100, // Needs to be large enough to touch each bucket.
+        max_reports: t.min_batch_size,
     };
 
     // Upload a number of reports (a few more than the aggregation rate).
     let mut rng = thread_rng();
-    let batch_interval = agg_info.batch_info.as_ref().unwrap();
-    for _ in 0..agg_info.agg_rate + 3 {
+    let batch_interval = t.batch_interval();
+    for _ in 0..agg_info.max_reports + 3 {
         let now = rng.gen_range(batch_interval.start..batch_interval.end());
         t.leader_post_expect_ok(
             &client,
@@ -109,10 +107,7 @@ async fn janus_helper() {
 
     // Aggregate first.
     let agg_telem = t.internal_process(&client, &agg_info).await;
-    assert_eq!(agg_telem.reports_processed, agg_info.agg_rate);
-    assert_eq!(agg_telem.reports_aggregated, agg_info.agg_rate);
-    let agg_telem = t.internal_process(&client, &agg_info).await;
-    assert_eq!(agg_telem.reports_aggregated, 3);
+    assert_eq!(agg_telem.reports_processed, agg_info.max_reports + 3);
     let agg_telem = t.internal_process(&client, &agg_info).await;
     assert_eq!(agg_telem.reports_aggregated, 0);
 
@@ -127,7 +122,7 @@ async fn janus_helper() {
         .await;
 
     let agg_telem = t.internal_process(&client, &agg_info).await;
-    assert_eq!(agg_telem.reports_collected, 13);
+    assert_eq!(agg_telem.reports_collected, 13, "reports collected");
 
     // Poll the collect URI before the ColleectResp is ready.
     let resp = client.get(collect_uri).send().await.unwrap();
diff --git a/daphne_worker_test/tests/test_runner.rs b/daphne_worker_test/tests/test_runner.rs
index fc03cbf..62c210e 100644
--- a/daphne_worker_test/tests/test_runner.rs
+++ b/daphne_worker_test/tests/test_runner.rs
@@ -134,7 +134,7 @@ pub struct TestRunner {
 impl TestRunner {
     pub async fn default() -> Self {
         let t = Self::with(DEFAULT_TASK, LEADER_TASK_LIST, HELPER_TASK_LIST).await;
-        t.internal_reset(&t.batch_info()).await;
+        t.internal_reset(&t.batch_interval()).await;
         t
     }
 
@@ -204,12 +204,12 @@ impl TestRunner {
             .unwrap()
     }
 
-    pub fn batch_info(&self) -> Option<Interval> {
+    pub fn batch_interval(&self) -> Interval {
         let start = self.now - (self.now % self.min_batch_duration);
-        Some(Interval {
+        Interval {
             start,
             duration: self.min_batch_duration * 2,
-        })
+        }
     }
 
     pub async fn get_hpke_configs(&self, client: &reqwest::Client) -> [HpkeConfig; 2] {
@@ -347,10 +347,10 @@ impl TestRunner {
     }
 
     #[allow(dead_code)]
-    pub async fn internal_reset(&self, batch_info: &Option<Interval>) {
+    pub async fn internal_reset(&self, batch_interval: &Interval) {
         let client = self.http_client();
-        post_internal_test_reset(&client, &self.leader_url, &self.task_id, batch_info).await;
-        post_internal_test_reset(&client, &self.helper_url, &self.task_id, batch_info).await;
+        post_internal_test_reset(&client, &self.leader_url, &self.task_id, batch_interval).await;
+        post_internal_test_reset(&client, &self.helper_url, &self.task_id, batch_interval).await;
     }
 }
 
@@ -375,8 +375,13 @@ impl JanusServerHandle {
 impl TestRunner {
     pub async fn janus_helper() -> (Self, JanusServerHandle) {
         let t = Self::with(JANUS_HELPER_TASK, LEADER_TASK_LIST, JANUS_HELPER_TASK_LIST).await;
-        post_internal_test_reset(&t.http_client(), &t.leader_url, &t.task_id, &t.batch_info())
-            .await;
+        post_internal_test_reset(
+            &t.http_client(),
+            &t.leader_url,
+            &t.task_id,
+            &t.batch_interval(),
+        )
+        .await;
 
         let task_id = janus_core::message::TaskId::get_decoded(t.task_id.as_ref()).unwrap();
         let aggregator_endpoints = vec![t.leader_url.clone(), t.helper_url.clone()];
@@ -517,7 +522,7 @@ async fn post_internal_test_reset(
     client: &reqwest::Client,
     base_url: &Url,
     task_id: &Id,
-    batch_info: &Option<Interval>,
+    batch_interval: &Interval,
 ) {
     let url = base_url
         .join(&format!(
@@ -528,7 +533,7 @@ async fn post_internal_test_reset(
 
     let req = client
         .post(url.as_str())
-        .body(serde_json::to_string(batch_info).unwrap());
+        .body(serde_json::to_string(batch_interval).unwrap());
     let resp = req.send().await.expect("request failed");
     assert_eq!(
         200,
diff --git a/daphne_worker_test/wrangler.toml b/daphne_worker_test/wrangler.toml
index a829f9c..229b3ad 100644
--- a/daphne_worker_test/wrangler.toml
+++ b/daphne_worker_test/wrangler.toml
@@ -25,6 +25,7 @@ bindings = [
   { name = "DAP_REPORT_STORE", class_name = "ReportStore" },
   { name = "DAP_AGGREGATE_STORE", class_name = "AggregateStore" },
   { name = "DAP_LEADER_STATE_STORE", class_name = "LeaderStateStore" },
+  { name = "DAP_LEADER_AGG_JOB_QUEUE", class_name = "LeaderAggregationJobQueue" },
   { name = "DAP_HELPER_STATE_STORE", class_name = "HelperStateStore" },
 ]