Repository: cloudflare/daphne
Pull request #129 (2022-10-03 16:09:27)

daphne: Add query type to task config
Based on #128 (merge that first).

Add the query type to the task configuration. For fixed-size queries it is necessary to handle batch sizes that exceed the maximum batch size, so do this as well.

While at it, change min_batch_duration to time_precision to match DAP-02.

cjpatton: > Pinning a reminder for myself to ask if we have tests for the InvalidBatchSIze error for chunky queries. I assume we'll get to that at some point later?

+1, I'm not sure we do. This should come up in #133.

Patch:
diff --git a/daphne/src/lib.rs b/daphne/src/lib.rs
index 5bfae0d..514af07 100644
--- a/daphne/src/lib.rs
+++ b/daphne/src/lib.rs
@@ -29,7 +29,7 @@
 //! on [`VdafConfig`](crate::VdafConfig) for producing reports and consuming aggregate results.
 
 use crate::{
-    messages::{CollectResp, HpkeConfig, Nonce, Time, TransitionFailure},
+    messages::{CollectResp, Duration, HpkeConfig, Nonce, Time, TransitionFailure},
     vdaf::{
         prio2::prio2_decode_prepare_state,
         prio3::{prio3_append_prepare_state, prio3_decode_prepare_state},
@@ -252,7 +252,7 @@ pub struct ProblemDetails {
 }
 
 /// DAP version used for a task.
-#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
 pub enum DapVersion {
     #[serde(rename = "v01")]
     Draft01,
@@ -304,18 +304,80 @@ pub struct DapGlobalConfig {
     pub supported_hpke_kems: Vec<HpkeKemId>,
 }
 
+/// DAP Query configuration.
+//
+// TODO(cjpatton) Once we implement maximum batch lifetime, put the parameter here.
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum DapQueryConfig {
+    /// The "time-interval" query type. Each report in the batch must fall into the time interval
+    /// specified by the query.
+    TimeInterval { min_batch_size: u64 },
+
+    /// The "fixed-size" query type. The Leader partitions the reports into arbitary batches of
+    /// roughly the same size.
+    FixedSize {
+        min_batch_size: u64,
+        max_batch_size: u64,
+    },
+}
+
+impl DapQueryConfig {
+    /// Check if the batch is ready to aggregate based on the report count. Returns an error if the
+    /// report count is too large.
+    pub(crate) fn check_batch_size(&self, report_count: u64) -> Result<bool, DapAbort> {
+        match self {
+            Self::TimeInterval { .. } => (),
+            Self::FixedSize {
+                min_batch_size: _,
+                max_batch_size,
+            } => {
+                if report_count > *max_batch_size {
+                    return Err(DapAbort::InvalidBatchSize);
+                }
+            }
+        };
+
+        Ok(report_count >= self.min_batch_size())
+    }
+
+    /// Return the minimum batch size determined by the query configuration.
+    pub fn min_batch_size(&self) -> u64 {
+        match self {
+            Self::TimeInterval { min_batch_size } => *min_batch_size,
+            Self::FixedSize { min_batch_size, .. } => *min_batch_size,
+        }
+    }
+}
+
 /// Per-task DAP parameters.
 #[derive(Deserialize, Serialize)]
 #[serde(try_from = "ShadowDapTaskConfig")]
 pub struct DapTaskConfig {
+    /// The protocol version (i.e., which draft).
     pub version: DapVersion,
+
+    /// Base URL of the Leader.
     pub leader_url: Url,
+
+    /// Base URL of the Helper.
     pub helper_url: Url,
-    pub min_batch_duration: u64, // seconds
-    pub min_batch_size: u64,     // number of reports
+
+    /// Report granularity. Used by the Client to truncate the timestamp and by the Aggregators to
+    /// constrain the batch interval of time=interval queries.
+    pub time_precision: Duration,
+
+    /// The query configuration for this task.
+    pub query: DapQueryConfig,
+
+    /// The VDAF configuration for this task.
     pub vdaf: VdafConfig,
+
+    /// VDAF verification key shared by the Aggregators. Used to aggregate reports.
     #[serde(skip_serializing)]
     pub(crate) vdaf_verify_key: VdafVerifyKey,
+
+    /// The Collector's HPKE configuration for this task.
     pub(crate) collector_hpke_config: HpkeConfig,
 }
 
@@ -324,11 +386,11 @@ impl DapTaskConfig {
     /// Convert at timestamp `now` into an [`Interval`] that contains it. The timestamp is the
     /// numbre of seconds since the beginning of UNIX time.
     pub fn query_for_current_batch_window(&self, now: u64) -> crate::messages::Query {
-        let start = now - (now % self.min_batch_duration);
+        let start = now - (now % self.time_precision);
         crate::messages::Query::TimeInterval {
             batch_interval: crate::messages::Interval {
                 start,
-                duration: self.min_batch_duration,
+                duration: self.time_precision,
             },
         }
     }
@@ -345,8 +407,8 @@ struct ShadowDapTaskConfig {
     version: DapVersion,
     leader_url: Url,
     helper_url: Url,
-    min_batch_duration: u64,
-    min_batch_size: u64,
+    time_precision: Duration,
+    query: DapQueryConfig,
     vdaf: VdafConfig,
     #[serde(with = "hex")]
     vdaf_verify_key: Vec<u8>,
@@ -365,8 +427,8 @@ impl TryFrom<ShadowDapTaskConfig> for DapTaskConfig {
             version: shadow.version,
             leader_url: shadow.leader_url,
             helper_url: shadow.helper_url,
-            min_batch_duration: shadow.min_batch_duration,
-            min_batch_size: shadow.min_batch_size,
+            time_precision: shadow.time_precision,
+            query: shadow.query,
             vdaf: shadow.vdaf,
             vdaf_verify_key,
             collector_hpke_config: shadow.collector_hpke_config,
diff --git a/daphne/src/messages.rs b/daphne/src/messages.rs
index a4d3b88..8c116a0 100644
--- a/daphne/src/messages.rs
+++ b/daphne/src/messages.rs
@@ -200,6 +200,8 @@ impl Decode for ReportShare {
     }
 }
 
+/// Batch parameter conveyed to the Helper by the Leader in the aggreagtion sub-protocol. Used to
+/// identify which batch the reports in the [`AggregateInitializeReq`] are intended for.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum BatchParameter {
     TimeInterval,
@@ -444,9 +446,9 @@ impl Interval {
 
     /// Check that the batch interval is valid for the given task configuration.
     pub fn is_valid_for(&self, task_config: &DapTaskConfig) -> bool {
-        if self.start % task_config.min_batch_duration != 0
-            || self.duration % task_config.min_batch_duration != 0
-            || self.duration < task_config.min_batch_duration
+        if self.start % task_config.time_precision != 0
+            || self.duration % task_config.time_precision != 0
+            || self.duration < task_config.time_precision
         {
             return false;
         }
diff --git a/daphne/src/roles.rs b/daphne/src/roles.rs
index 01f354e..600bd9f 100644
--- a/daphne/src/roles.rs
+++ b/daphne/src/roles.rs
@@ -424,8 +424,13 @@ pub trait DapLeader<'a, S>: DapAuthorizedSender<S> + DapAggregator<'a, S> {
             .get_agg_share(&collect_req.task_id, &collect_req.query)
             .await?;
 
-        // Check that the minimum batch size is met.
-        if leader_agg_share.report_count < task_config.min_batch_size {
+        // Check the batch size. If not not ready, then return early.
+        //
+        // TODO Consider logging this error, as it shouuld never happen.
+        if !task_config
+            .query
+            .check_batch_size(leader_agg_share.report_count)?
+        {
             return Ok(0);
         }
 
@@ -738,8 +743,12 @@ pub trait DapHelper<'a, S>: DapAggregator<'a, S> {
             return Err(DapAbort::BatchMismatch);
         }
 
-        // Check that the minimum batch size is met.
-        if agg_share.report_count < task_config.min_batch_size {
+        // Check the batch size.
+        if !task_config
+            .query
+            .check_batch_size(agg_share.report_count)
+            .unwrap_or(false)
+        {
             return Err(DapAbort::InvalidBatchSize);
         }
 
diff --git a/daphne/src/vdaf/mod.rs b/daphne/src/vdaf/mod.rs
index bed30ad..6bb44ba 100644
--- a/daphne/src/vdaf/mod.rs
+++ b/daphne/src/vdaf/mod.rs
@@ -361,7 +361,7 @@ impl VdafConfig {
                 task_id: task_id.clone(),
                 agg_job_id: agg_job_id.clone(),
                 agg_param: Vec::default(),
-                batch_param: BatchParameter::TimeInterval,
+                batch_param: BatchParameter::TimeInterval, // TODO(issue #100)
                 report_shares: seq,
             },
         ))
diff --git a/daphne_worker/src/config.rs b/daphne_worker/src/config.rs
index 77be5a1..ae27b10 100644
--- a/daphne_worker/src/config.rs
+++ b/daphne_worker/src/config.rs
@@ -205,7 +205,7 @@ impl<D> DaphneWorkerConfig<D> {
         let mut bucket_seed = [0; 8];
         PrgAes128::seed_stream(&self.bucket_key, metadata.nonce.as_ref()).fill(&mut bucket_seed);
         let bucket = u64::from_be_bytes(bucket_seed) % self.bucket_count;
-        let time = metadata.time - (metadata.time % task_config.min_batch_duration);
+        let time = metadata.time - (metadata.time % task_config.time_precision);
         durable_report_store_name(&task_config.version, &task_id.to_hex(), time, bucket)
     }
 
@@ -224,7 +224,7 @@ impl<D> DaphneWorkerConfig<D> {
             .tasks
             .get(task_id)
             .ok_or_else(|| Error::RustError(format!("Unrecognized task ID: {}", task_id_hex)))?
-            .min_batch_duration;
+            .time_precision;
 
         if interval.end() <= interval.start {
             return Err(Error::RustError(
@@ -235,12 +235,12 @@ impl<D> DaphneWorkerConfig<D> {
 
         if interval.start % time_step != 0 || interval.end() % time_step != 0 {
             return Err(Error::RustError(
-                "Batch interval does not align with min_batch_duration".to_string(),
+                "Batch interval does not align with time_precision".to_string(),
             ));
         }
 
         Ok(DurableNameIterator {
-            version: version.clone(),
+            version: *version,
             task_id_hex,
             time_start: interval.start,
             time_mod: interval.end() - interval.start,
diff --git a/daphne_worker/src/dap.rs b/daphne_worker/src/dap.rs
index 0b46485..9e4e7ef 100644
--- a/daphne_worker/src/dap.rs
+++ b/daphne_worker/src/dap.rs
@@ -252,7 +252,7 @@ impl<'a, D> DapAggregator<'a, BearerToken> for DaphneWorkerConfig<D> {
         let durable = self.durable();
         let mut requests = Vec::new();
         for window in (batch_interval.start..batch_interval.end())
-            .step_by(task_config.min_batch_duration.try_into().unwrap())
+            .step_by(task_config.time_precision.try_into().unwrap())
         {
             let durable_name =
                 durable_agg_store_name(&task_config.version, &task_id.to_hex(), window);
@@ -288,7 +288,7 @@ impl<'a, D> DapAggregator<'a, BearerToken> for DaphneWorkerConfig<D> {
         let task_config = self.try_get_task_config_for(task_id)?;
 
         let agg_shares =
-            DapAggregateShare::batches_from_out_shares(out_shares, task_config.min_batch_duration)?;
+            DapAggregateShare::batches_from_out_shares(out_shares, task_config.time_precision)?;
 
         let durable = self.durable();
         let mut requests = Vec::new();
@@ -317,7 +317,7 @@ impl<'a, D> DapAggregator<'a, BearerToken> for DaphneWorkerConfig<D> {
         let durable = self.durable();
         let mut requests = Vec::new();
         for window in (batch_interval.start..batch_interval.end())
-            .step_by(task_config.min_batch_duration.try_into().unwrap())
+            .step_by(task_config.time_precision.try_into().unwrap())
         {
             let durable_name =
                 durable_agg_store_name(&task_config.version, &task_id.to_hex(), window);
@@ -375,7 +375,7 @@ impl<'a, D> DapAggregator<'a, BearerToken> for DaphneWorkerConfig<D> {
         let durable = self.durable();
         let mut requests = Vec::new();
         for window in (batch_interval.start..batch_interval.end())
-            .step_by(task_config.min_batch_duration.try_into().unwrap())
+            .step_by(task_config.time_precision.try_into().unwrap())
         {
             let durable_name =
                 durable_agg_store_name(&task_config.version, &task_id.to_hex(), window);


Test patch:
diff --git a/daphne/src/roles_test.rs b/daphne/src/roles_test.rs
index 41a25f3..39077f5 100644
--- a/daphne/src/roles_test.rs
+++ b/daphne/src/roles_test.rs
@@ -19,8 +19,8 @@ use crate::{
         AggStoreState, BucketInfo, MockAggregateInfo, MockAggregator, ReportStore,
         COLLECTOR_BEARER_TOKEN, HPKE_RECEIVER_CONFIG_LIST, LEADER_BEARER_TOKEN,
     },
-    DapAbort, DapAggregateShare, DapCollectJob, DapLeaderTransition, DapMeasurement, DapRequest,
-    DapTaskConfig, DapVersion, Prio3Config, VdafConfig,
+    DapAbort, DapAggregateShare, DapCollectJob, DapLeaderTransition, DapMeasurement,
+    DapQueryConfig, DapRequest, DapTaskConfig, DapVersion, Prio3Config, VdafConfig,
 };
 use assert_matches::assert_matches;
 use matchit::Router;
@@ -63,7 +63,10 @@ impl MockAggregator {
         let task_id = self.nominal_task_id();
         let task_config = self.tasks.get(task_id).unwrap();
         let version = task_config.version.clone();
-        let batch_param = BatchParameter::TimeInterval;
+        let batch_param = match task_config.query {
+            DapQueryConfig::TimeInterval { .. } => BatchParameter::TimeInterval,
+            _ => panic!("TODO(issue #100)"),
+        };
 
         DapRequest {
             version,
@@ -772,8 +775,8 @@ async fn http_post_collect_fail_invalid_batch_interval() {
         task_id: task_id.clone(),
         query: Query::TimeInterval {
             batch_interval: Interval {
-                start: now - (now % task_config.min_batch_duration),
-                duration: leader.global_config.max_batch_duration + task_config.min_batch_duration,
+                start: now - (now % task_config.time_precision),
+                duration: leader.global_config.max_batch_duration + task_config.time_precision,
             },
         },
         agg_param: Vec::default(),
@@ -799,10 +802,10 @@ async fn http_post_collect_fail_invalid_batch_interval() {
         query: Query::TimeInterval {
             batch_interval: Interval {
                 start: now
-                    - (now % task_config.min_batch_duration)
+                    - (now % task_config.time_precision)
                     - leader.global_config.min_batch_interval_start
-                    - task_config.min_batch_duration,
-                duration: task_config.min_batch_duration * 2,
+                    - task_config.time_precision,
+                duration: task_config.time_precision * 2,
             },
         },
         agg_param: Vec::default(),
@@ -827,10 +830,10 @@ async fn http_post_collect_fail_invalid_batch_interval() {
         task_id: task_id.clone(),
         query: Query::TimeInterval {
             batch_interval: Interval {
-                start: now - (now % task_config.min_batch_duration)
+                start: now - (now % task_config.time_precision)
                     + leader.global_config.max_batch_interval_end
-                    - task_config.min_batch_duration,
-                duration: task_config.min_batch_duration * 2,
+                    - task_config.time_precision,
+                duration: task_config.time_precision * 2,
             },
         },
         agg_param: Vec::default(),
@@ -864,7 +867,7 @@ async fn http_post_collect_succeed_max_batch_interval() {
         query: Query::TimeInterval {
             batch_interval: Interval {
                 start: now
-                    - (now % task_config.min_batch_duration)
+                    - (now % task_config.time_precision)
                     - leader.global_config.max_batch_duration / 2,
                 duration: leader.global_config.max_batch_duration,
             },
diff --git a/daphne/src/testing.rs b/daphne/src/testing.rs
index be4e9bc..115fc7f 100644
--- a/daphne/src/testing.rs
+++ b/daphne/src/testing.rs
@@ -50,8 +50,12 @@ pub const TASK_LIST: &str = r#"{
             "aead_id": "Aes128Gcm",
             "public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"
         },
-        "min_batch_duration": 3600,
-        "min_batch_size": 1,
+        "time_precision": 3600,
+        "query": {
+            "time_interval": {
+                "min_batch_size": 1
+            }
+        },
         "vdaf": {
             "prio3": "count"
         },
@@ -309,7 +313,7 @@ impl<'a> DapAggregator<'a, BearerToken> for MockAggregator {
             .ok_or_else(|| DapError::fatal("task not found"))?;
 
         let agg_shares =
-            DapAggregateShare::batches_from_out_shares(out_shares, task_config.min_batch_duration)?;
+            DapAggregateShare::batches_from_out_shares(out_shares, task_config.time_precision)?;
 
         let mut agg_store_mutex_guard = self
             .agg_store
@@ -739,7 +743,7 @@ pub(crate) struct BucketInfo {
 
 impl BucketInfo {
     pub(crate) fn new(task_config: &DapTaskConfig, task_id: &Id, time: Time) -> Self {
-        let window = time - (time % task_config.min_batch_duration);
+        let window = time - (time % task_config.time_precision);
 
         Self {
             task_id: task_id.clone(),
diff --git a/daphne_worker/src/config_test.rs b/daphne_worker/src/config_test.rs
index 8fd2782..3b69cc8 100644
--- a/daphne_worker/src/config_test.rs
+++ b/daphne_worker/src/config_test.rs
@@ -23,8 +23,12 @@ const DAP_TASK_LIST: &str = r#"{
             "aead_id": "Aes128Gcm",
             "public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"
         },
-        "min_batch_duration": 3600,
-        "min_batch_size": 100,
+        "time_precision": 3600,
+        "query": {
+            "time_interval": {
+                "min_batch_size": 100
+            }
+        },
         "vdaf": {
             "prio3": {
                 "sum": {
@@ -45,8 +49,13 @@ const DAP_TASK_LIST: &str = r#"{
             "aead_id": "Aes128Gcm",
             "public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"
         },
-        "min_batch_duration": 3600,
-        "min_batch_size": 10,
+        "time_precision": 3600,
+        "query": {
+            "fixed_size": {
+                "min_batch_size": 100,
+                "max_batch_size": 110
+            }
+        },
         "vdaf": {
             "prio2": {
                 "dimension": 100
diff --git a/daphne_worker_test/helper.env b/daphne_worker_test/helper.env
index 58a7b0f..3b4cfc1 100644
--- a/daphne_worker_test/helper.env
+++ b/daphne_worker_test/helper.env
@@ -19,6 +19,6 @@ DAP_HELPER_STATE_STORE_GARBAGE_COLLECT_AFTER_SECS = '10'
 #
 # NOTE(nakatsuka-y) The leader_url and helper_url must end with a "/".
 # When adding paths, they must not start with a "/".
-DAP_TASK_LIST = '{"f285be3caf948fcfc36b7d32181c14db95c55f04f55a2db2ee439c5879264e1f":{"version":"v01","leader_url":"http://leader:8787/v01/","helper_url":"http://helper:8788/v01/","collector_hpke_config":{"id":23,"kem_id":"X25519HkdfSha256","kdf_id":"HkdfSha256","aead_id":"Aes128Gcm","public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"},"min_batch_duration":3600,"min_batch_size":10,"vdaf":{"prio3":{"sum":{"bits":10}}},"vdaf_verify_key":"1fd8d30dc0e0b7ac81f0050fcab0782d"}}'
+DAP_TASK_LIST = '{"f285be3caf948fcfc36b7d32181c14db95c55f04f55a2db2ee439c5879264e1f":{"version":"v01","leader_url":"http://leader:8787/v01/","helper_url":"http://helper:8788/v01/","collector_hpke_config":{"id":23,"kem_id":"X25519HkdfSha256","kdf_id":"HkdfSha256","aead_id":"Aes128Gcm","public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"},"time_precision":3600,"query":{"time_interval":{"min_batch_size":10}},"vdaf":{"prio3":{"sum":{"bits":10}}},"vdaf_verify_key":"1fd8d30dc0e0b7ac81f0050fcab0782d"}}'
 
 DAP_LEADER_BEARER_TOKEN_LIST = '{"f285be3caf948fcfc36b7d32181c14db95c55f04f55a2db2ee439c5879264e1f":"This is an auth token ivA1e7LpnySDNn1AulaZggFLQ1n7jZ8GWOUO7GY4hgs="}'
diff --git a/daphne_worker_test/leader.env b/daphne_worker_test/leader.env
index d085a2b..c2bb126 100644
--- a/daphne_worker_test/leader.env
+++ b/daphne_worker_test/leader.env
@@ -19,7 +19,7 @@ DAP_COLLECT_ID_KEY = "b416a85d280591d6da14e5b75a7d6e31"
 #
 # NOTE(nakatsuka-y) The leader_url and helper_url must end with a "/".
 # When adding paths, they must not start with a "/".
-DAP_TASK_LIST = '{"f285be3caf948fcfc36b7d32181c14db95c55f04f55a2db2ee439c5879264e1f":{"version":"v01","leader_url":"http://leader:8787/v01/","helper_url":"http://helper:8788/v01/","collector_hpke_config":{"id":23,"kem_id":"X25519HkdfSha256","kdf_id":"HkdfSha256","aead_id":"Aes128Gcm","public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"},"min_batch_duration":3600,"min_batch_size":10,"vdaf":{"prio3":{"sum":{"bits":10}}},"vdaf_verify_key":"1fd8d30dc0e0b7ac81f0050fcab0782d"},"410d5e0abd94a88b8435a192cc458cc1667da2989827584cbf8a591626d5a61f":{"version":"v01","leader_url":"http://leader:8787/v01/","helper_url":"http://127.0.0.1:9788/","collector_hpke_config":{"id":23,"kem_id":"X25519HkdfSha256","kdf_id":"HkdfSha256","aead_id":"Aes128Gcm","public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"},"min_batch_duration":3600,"min_batch_size":10,"vdaf":{"prio3":{"sum":{"bits":10}}},"vdaf_verify_key":"01d6232e33fe7e63b4531e3706efa8cc"}}'
+DAP_TASK_LIST = '{"f285be3caf948fcfc36b7d32181c14db95c55f04f55a2db2ee439c5879264e1f":{"version":"v01","leader_url":"http://leader:8787/v01/","helper_url":"http://helper:8788/v01/","collector_hpke_config":{"id":23,"kem_id":"X25519HkdfSha256","kdf_id":"HkdfSha256","aead_id":"Aes128Gcm","public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"},"time_precision":3600,"query":{"time_interval":{"min_batch_size":10}},"vdaf":{"prio3":{"sum":{"bits":10}}},"vdaf_verify_key":"1fd8d30dc0e0b7ac81f0050fcab0782d"},"410d5e0abd94a88b8435a192cc458cc1667da2989827584cbf8a591626d5a61f":{"version":"v01","leader_url":"http://leader:8787/v01/","helper_url":"http://127.0.0.1:9788/","collector_hpke_config":{"id":23,"kem_id":"X25519HkdfSha256","kdf_id":"HkdfSha256","aead_id":"Aes128Gcm","public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"},"time_precision":3600,"query":{"time_interval":{"min_batch_size":10}},"vdaf":{"prio3":{"sum":{"bits":10}}},"vdaf_verify_key":"01d6232e33fe7e63b4531e3706efa8cc"}}'
 
 DAP_LEADER_BEARER_TOKEN_LIST = '{"f285be3caf948fcfc36b7d32181c14db95c55f04f55a2db2ee439c5879264e1f":"This is an auth token ivA1e7LpnySDNn1AulaZggFLQ1n7jZ8GWOUO7GY4hgs=","410d5e0abd94a88b8435a192cc458cc1667da2989827584cbf8a591626d5a61f":"This is a differnt token 72938088f14b7ef318ef42ba72395a22"}'
 
diff --git a/daphne_worker_test/tests/e2e.rs b/daphne_worker_test/tests/e2e.rs
index 5186f30..b5bd731 100644
--- a/daphne_worker_test/tests/e2e.rs
+++ b/daphne_worker_test/tests/e2e.rs
@@ -195,7 +195,7 @@ async fn e2e_internal_leader_process() {
 
     let agg_info = InternalAggregateInfo {
         max_buckets: 100, // Needs to be sufficiently large to touch each bucket.
-        max_reports: t.min_batch_size,
+        max_reports: t.query_config.min_batch_size(),
     };
 
     let batch_interval = t.batch_interval();
@@ -288,7 +288,7 @@ async fn e2e_leader_collect_ok() {
 
     // The reports are uploaded in the background.
     let mut rng = thread_rng();
-    for _ in 0..t.min_batch_size {
+    for _ in 0..t.query_config.min_batch_size() {
         let now = rng.gen_range(batch_interval.start..batch_interval.end());
         t.leader_post_expect_ok(
             &client,
@@ -330,15 +330,18 @@ async fn e2e_leader_collect_ok() {
         )
         .await;
     assert_eq!(
-        agg_telem.reports_processed, t.min_batch_size,
+        agg_telem.reports_processed,
+        t.query_config.min_batch_size(),
         "reports processed"
     );
     assert_eq!(
-        agg_telem.reports_aggregated, t.min_batch_size,
+        agg_telem.reports_aggregated,
+        t.query_config.min_batch_size(),
         "reports aggregated"
     );
     assert_eq!(
-        agg_telem.reports_collected, t.min_batch_size,
+        agg_telem.reports_collected,
+        t.query_config.min_batch_size(),
         "reports collected"
     );
 
@@ -362,7 +365,10 @@ async fn e2e_leader_collect_ok() {
         )
         .await
         .unwrap();
-    assert_eq!(agg_res, DapAggregateResult::U128(t.min_batch_size as u128));
+    assert_eq!(
+        agg_res,
+        DapAggregateResult::U128(t.query_config.min_batch_size() as u128)
+    );
 
     // Poll the collect URI once more. Expect the response to be the same as the first, per HTTP
     // GET semantics.
@@ -399,7 +405,7 @@ async fn e2e_leader_collect_ok_interleaved() {
 
     // The reports are uploaded in the background.
     let mut rng = thread_rng();
-    for _ in 0..t.min_batch_size {
+    for _ in 0..t.query_config.min_batch_size() {
         let now = rng.gen_range(batch_interval.start..batch_interval.end());
         t.leader_post_expect_ok(
             &client,
@@ -421,7 +427,8 @@ async fn e2e_leader_collect_ok_interleaved() {
     // 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,
+        agg_telem.reports_processed,
+        t.query_config.min_batch_size(),
         "reports processed"
     );
 
@@ -440,7 +447,8 @@ async fn e2e_leader_collect_ok_interleaved() {
     // ... 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,
+        agg_telem.reports_collected,
+        t.query_config.min_batch_size(),
         "reports collected"
     );
 }
@@ -455,7 +463,7 @@ async fn e2e_leader_collect_not_ready_min_batch_size() {
 
     // A number of reports are uploaded, but not enough to meet the minimum batch requirement.
     let mut rng = thread_rng();
-    for _ in 0..t.min_batch_size - 1 {
+    for _ in 0..t.query_config.min_batch_size() - 1 {
         let now = rng.gen_range(batch_interval.start..batch_interval.end());
         t.leader_post_expect_ok(
             &client,
@@ -492,8 +500,14 @@ async fn e2e_leader_collect_not_ready_min_batch_size() {
             },
         )
         .await;
-    assert_eq!(agg_telem.reports_processed, t.min_batch_size - 1);
-    assert_eq!(agg_telem.reports_aggregated, t.min_batch_size - 1);
+    assert_eq!(
+        agg_telem.reports_processed,
+        t.query_config.min_batch_size() - 1
+    );
+    assert_eq!(
+        agg_telem.reports_aggregated,
+        t.query_config.min_batch_size() - 1
+    );
     assert_eq!(agg_telem.reports_collected, 0);
 
     // Poll the collect URI before the ColleectResp is ready.
@@ -527,7 +541,7 @@ async fn e2e_leader_collect_accept_max_batch_duration() {
     let t = TestRunner::default().await;
     let client = t.http_client();
     let batch_interval = Interval {
-        start: t.now - (t.now % t.min_batch_duration) - t.max_batch_duration / 2,
+        start: t.now - (t.now % t.time_precision) - t.max_batch_duration / 2,
         duration: t.max_batch_duration,
     };
 
@@ -550,7 +564,7 @@ async fn e2e_leader_collect_abort_invalid_batch_interval() {
     let batch_interval = t.batch_interval();
     let path = "collect";
 
-    // Start of batch interval does not align with min_batch_duration.
+    // Start of batch interval does not align with time_precision.
     let collect_req = CollectReq {
         task_id: t.task_id.clone(),
         query: Query::TimeInterval {
@@ -605,7 +619,7 @@ async fn e2e_leader_collect_abort_overlapping_batch_interval() {
 
     // The reports are uploaded in the background.
     let mut rng = thread_rng();
-    for _ in 0..t.min_batch_size {
+    for _ in 0..t.query_config.min_batch_size() {
         let now = rng.gen_range(batch_interval.start..batch_interval.end());
         t.leader_post_expect_ok(
             &client,
@@ -642,15 +656,18 @@ async fn e2e_leader_collect_abort_overlapping_batch_interval() {
         )
         .await;
     assert_eq!(
-        agg_telem.reports_processed, t.min_batch_size,
+        agg_telem.reports_processed,
+        t.query_config.min_batch_size(),
         "reports processed"
     );
     assert_eq!(
-        agg_telem.reports_aggregated, t.min_batch_size,
+        agg_telem.reports_aggregated,
+        t.query_config.min_batch_size(),
         "reports aggregated"
     );
     assert_eq!(
-        agg_telem.reports_collected, t.min_batch_size,
+        agg_telem.reports_collected,
+        t.query_config.min_batch_size(),
         "reports collected"
     );
 
diff --git a/daphne_worker_test/tests/janus.rs b/daphne_worker_test/tests/janus.rs
index 4dcba21..76d21fd 100644
--- a/daphne_worker_test/tests/janus.rs
+++ b/daphne_worker_test/tests/janus.rs
@@ -39,7 +39,7 @@ async fn janus_client() {
     let janus_client_parameters = janus_client::ClientParameters::new(
         task_id,
         vec![t.leader_url.clone(), t.helper_url.clone()],
-        janus_core::message::Duration::from_seconds(t.min_batch_duration),
+        janus_core::message::Duration::from_seconds(t.time_precision),
     );
 
     let client_clock = janus_core::time::test_util::MockClock::new(
@@ -78,7 +78,7 @@ async fn janus_helper() {
     let hpke_config_list = t.get_hpke_configs(&client).await;
     let agg_info = InternalAggregateInfo {
         max_buckets: 100, // Needs to be large enough to touch each bucket.
-        max_reports: t.min_batch_size,
+        max_reports: t.query_config.min_batch_size(),
     };
 
     // Upload a number of reports (a few more than the aggregation rate).
diff --git a/daphne_worker_test/tests/test_runner.rs b/daphne_worker_test/tests/test_runner.rs
index a529b1c..297902b 100644
--- a/daphne_worker_test/tests/test_runner.rs
+++ b/daphne_worker_test/tests/test_runner.rs
@@ -7,7 +7,8 @@ use assert_matches::assert_matches;
 use daphne::{
     constants::MEDIA_TYPE_COLLECT_REQ,
     messages::{HpkeConfig, Id, Interval},
-    DapGlobalConfig, DapLeaderProcessTelemetry, DapTaskConfig, DapVersion, VdafConfig,
+    DapGlobalConfig, DapLeaderProcessTelemetry, DapQueryConfig, DapTaskConfig, DapVersion,
+    VdafConfig,
 };
 use daphne_worker::InternalAggregateInfo;
 use futures::channel::oneshot::Sender;
@@ -53,8 +54,12 @@ const LEADER_TASK_LIST: &str = r#"{
             "aead_id": "Aes128Gcm",
             "public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"
         },
-        "min_batch_duration": 3600,
-        "min_batch_size": 10,
+        "time_precision": 3600,
+        "query": {
+            "time_interval": {
+                "min_batch_size": 10
+            }
+        },
         "vdaf": {
             "prio3": {
                 "sum": {
@@ -75,8 +80,12 @@ const LEADER_TASK_LIST: &str = r#"{
             "aead_id": "Aes128Gcm",
             "public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"
         },
-        "min_batch_duration": 3600,
-        "min_batch_size": 10,
+        "time_precision": 3600,
+        "query": {
+            "time_interval": {
+                "min_batch_size": 10
+            }
+        },
         "vdaf": {
             "prio3": {
                 "sum": {
@@ -106,8 +115,12 @@ const HELPER_TASK_LIST: &str = r#"{
             "aead_id": "Aes128Gcm",
             "public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"
         },
-        "min_batch_duration": 3600,
-        "min_batch_size": 10,
+        "time_precision": 3600,
+        "query": {
+            "time_interval": {
+                "min_batch_size": 10
+            }
+        },
         "vdaf": {
             "prio3": {
                 "sum": {
@@ -131,8 +144,12 @@ pub(crate) const JANUS_HELPER_TASK_LIST: &str = r#"{
             "aead_id": "Aes128Gcm",
             "public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"
         },
-        "min_batch_duration": 3600,
-        "min_batch_size": 10,
+        "time_precision": 3600,
+        "query": {
+            "time_interval": {
+                "min_batch_size": 10
+            }
+        },
         "vdaf": {
             "prio3": {
                 "sum": {
@@ -167,8 +184,8 @@ pub struct TestRunner {
     pub version: DapVersion,
     pub task_id: Id,
     pub now: u64,
-    pub min_batch_duration: u64,
-    pub min_batch_size: u64,
+    pub time_precision: u64,
+    pub query_config: DapQueryConfig,
     pub max_batch_duration: u64,
     pub min_batch_interval_start: u64,
     pub max_batch_interval_end: u64,
@@ -231,8 +248,8 @@ impl TestRunner {
             version: task_config.version.clone(),
             task_id: task_id.clone(),
             now,
-            min_batch_duration: task_config.min_batch_duration,
-            min_batch_size: task_config.min_batch_size,
+            time_precision: task_config.time_precision,
+            query_config: task_config.query.clone(),
             max_batch_duration: global_config.max_batch_duration,
             min_batch_interval_start: global_config.min_batch_interval_start,
             max_batch_interval_end: global_config.max_batch_interval_end,
@@ -244,13 +261,10 @@ impl TestRunner {
         // Ensure the helper has a matching task config.
         if let Some(helper_task_config) = helper_task_list.get(&task_id) {
             assert_eq!(
-                helper_task_config.min_batch_duration,
-                task_config.min_batch_duration
-            );
-            assert_eq!(
-                helper_task_config.min_batch_size,
-                task_config.min_batch_size
+                helper_task_config.time_precision,
+                task_config.time_precision
             );
+            assert_eq!(helper_task_config.query, task_config.query);
             assert_eq!(helper_task_config.vdaf, task_config.vdaf);
         } else {
             panic!("Helper does not have as matching task configuration");
@@ -271,10 +285,10 @@ impl TestRunner {
     }
 
     pub fn batch_interval(&self) -> Interval {
-        let start = self.now - (self.now % self.min_batch_duration);
+        let start = self.now - (self.now % self.time_precision);
         Interval {
             start,
-            duration: self.min_batch_duration * 2,
+            duration: self.time_precision * 2,
         }
     }
 
@@ -500,8 +514,8 @@ impl TestRunner {
             janus_core::message::Role::Helper,
             vec![vdaf_verify_key],
             1, // max_batch_lifetime
-            t.min_batch_size,
-            janus_core::message::Duration::from_seconds(t.min_batch_duration),
+            t.query_config.min_batch_size(),
+            janus_core::message::Duration::from_seconds(t.time_precision),
             janus_core::message::Duration::from_seconds(0), // clock skew tolerance
             collector_hpke_config,
             vec![leader_bearer_token],