Repository: cloudflare/daphne
Pull request #730 (2024-11-28 12:13:49)

Rename fixed-size queries to leader-selected queries
Draft-12 renames the "fixed-size" query type to "leader-selected". This PR makes this change in our codebase, rather than leaving the old names in place.
Even though we still have draft-09 code in place I renamed it here to keep the code consistent, given that this has no effect on the wire.


Patch:
diff --git a/crates/dapf/src/acceptance/mod.rs b/crates/dapf/src/acceptance/mod.rs
index 4edf036..bee2dcd 100644
--- a/crates/dapf/src/acceptance/mod.rs
+++ b/crates/dapf/src/acceptance/mod.rs
@@ -380,7 +380,7 @@ impl Test {
             time_precision: 3600,
             lifetime: 60,
             min_batch_size: reports_per_batch.try_into().unwrap(),
-            query: DapQueryConfig::FixedSize {
+            query: DapQueryConfig::LeaderSelected {
                 max_batch_size: NonZeroU32::new(reports_per_batch.try_into().unwrap()),
             },
             vdaf: self.vdaf_config,
@@ -554,7 +554,7 @@ impl Test {
     ) -> Result<Duration> {
         // Prepare AggregateShareReq.
         let agg_share_req = AggregateShareReq {
-            batch_sel: BatchSelector::FixedSizeByBatchId { batch_id },
+            batch_sel: BatchSelector::LeaderSelectedByBatchId { batch_id },
             agg_param: Vec::new(),
             report_count: leader_agg_share.report_count,
             checksum: leader_agg_share.checksum,
@@ -636,7 +636,7 @@ impl Test {
         ////
 
         let batch_id = BatchId(rngs::OsRng.gen());
-        let part_batch_sel = PartialBatchSelector::FixedSizeByBatchId { batch_id };
+        let part_batch_sel = PartialBatchSelector::LeaderSelectedByBatchId { batch_id };
 
         let (leader_agg_share, agg_job_duration) = self
             .run_agg_jobs(
diff --git a/crates/dapf/src/cli_parsers.rs b/crates/dapf/src/cli_parsers.rs
index 16666df..bec3b6e 100644
--- a/crates/dapf/src/cli_parsers.rs
+++ b/crates/dapf/src/cli_parsers.rs
@@ -153,8 +153,8 @@ impl FromStr for CliDapQueryConfig {
     fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
         if s == "time-interval" {
             Ok(Self(DapQueryConfig::TimeInterval))
-        } else if let Some(size) = s.strip_prefix("fixed-size") {
-            Ok(Self(DapQueryConfig::FixedSize {
+        } else if let Some(size) = s.strip_prefix("leader-selected") {
+            Ok(Self(DapQueryConfig::LeaderSelected {
                 max_batch_size: if let Some(size) = size.strip_prefix("-") {
                     Some(
                         size.parse()
@@ -163,7 +163,9 @@ impl FromStr for CliDapQueryConfig {
                 } else if size.is_empty() {
                     None
                 } else {
-                    return Err(format!("{size} is an invalid fixed size max batch size"));
+                    return Err(format!(
+                        "{size} is an invalid leader selected max batch size"
+                    ));
                 },
             }))
         } else {
diff --git a/crates/dapf/src/main.rs b/crates/dapf/src/main.rs
index 31f7316..361899f 100644
--- a/crates/dapf/src/main.rs
+++ b/crates/dapf/src/main.rs
@@ -771,8 +771,8 @@ async fn handle_decode_actions(action: DecodeAction) -> anyhow::Result<()> {
                     let batch_selector = batch_selector.unwrap_or_else(||
                         match message.part_batch_sel {
                             PartialBatchSelector::TimeInterval => panic!("can't deduce the time interval, please provide a batch selector"),
-                            PartialBatchSelector::FixedSizeByBatchId { batch_id } => {
-                                BatchSelector::FixedSizeByBatchId { batch_id }
+                            PartialBatchSelector::LeaderSelectedByBatchId { batch_id } => {
+                                BatchSelector::LeaderSelectedByBatchId { batch_id }
                             }
                         }
                     );
@@ -844,7 +844,7 @@ async fn handle_test_routes(action: TestAction, http_client: HttpClient) -> anyh
             let vdaf_verify_key = encode_base64url(vdaf.gen_verify_key());
             let CliDapQueryConfig(query) = use_or_request_from_user_or_default(
                 query,
-                || DapQueryConfig::FixedSize {
+                || DapQueryConfig::LeaderSelected {
                     max_batch_size: None,
                 },
                 "query",
@@ -876,7 +876,7 @@ async fn handle_test_routes(action: TestAction, http_client: HttpClient) -> anyh
                 vdaf_verify_key,
                 query_type: match query {
                     DapQueryConfig::TimeInterval => 1,
-                    DapQueryConfig::FixedSize { .. } => 2,
+                    DapQueryConfig::LeaderSelected { .. } => 2,
                 },
                 min_batch_size: use_or_request_from_user_or_default(
                     min_batch_size,
@@ -885,7 +885,7 @@ async fn handle_test_routes(action: TestAction, http_client: HttpClient) -> anyh
                 )?,
                 max_batch_size: match query {
                     DapQueryConfig::TimeInterval => None,
-                    DapQueryConfig::FixedSize { max_batch_size } => max_batch_size,
+                    DapQueryConfig::LeaderSelected { max_batch_size } => max_batch_size,
                 },
                 time_precision: use_or_request_from_user_or_default(
                     time_precision,
diff --git a/crates/daphne-server/src/roles/aggregator.rs b/crates/daphne-server/src/roles/aggregator.rs
index f00e9d8..c7774a1 100644
--- a/crates/daphne-server/src/roles/aggregator.rs
+++ b/crates/daphne-server/src/roles/aggregator.rs
@@ -294,7 +294,7 @@ impl DapAggregator for crate::App {
             }))?;
         let version = task_config.as_ref().version;
 
-        let agg_span = task_config.batch_span_for_sel(&BatchSelector::FixedSizeByBatchId {
+        let agg_span = task_config.batch_span_for_sel(&BatchSelector::LeaderSelectedByBatchId {
             batch_id: *batch_id,
         })?;
 
diff --git a/crates/daphne-server/src/roles/mod.rs b/crates/daphne-server/src/roles/mod.rs
index 397c2fa..2520684 100644
--- a/crates/daphne-server/src/roles/mod.rs
+++ b/crates/daphne-server/src/roles/mod.rs
@@ -283,7 +283,7 @@ mod test_utils {
                         err = "command failed: unexpected max batch size"
                     ))
                 }
-                (2, max_batch_size) => DapQueryConfig::FixedSize { max_batch_size },
+                (2, max_batch_size) => DapQueryConfig::LeaderSelected { max_batch_size },
                 _ => {
                     return Err(fatal_error!(
                         err = "command failed: unrecognized query type"
diff --git a/crates/daphne-service-utils/src/durable_requests/bindings/mod.rs b/crates/daphne-service-utils/src/durable_requests/bindings/mod.rs
index fd45b59..3807e9f 100644
--- a/crates/daphne-service-utils/src/durable_requests/bindings/mod.rs
+++ b/crates/daphne-service-utils/src/durable_requests/bindings/mod.rs
@@ -144,7 +144,7 @@ mod tests {
             "batch/IiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiI",
             format!(
                 "{}",
-                DapBatchBucket::FixedSize {
+                DapBatchBucket::LeaderSelected {
                     batch_id: BatchId([34; 32]),
                     shard: 0,
                 }
@@ -164,7 +164,7 @@ mod tests {
             "batch/IiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiI/shard/2323",
             format!(
                 "{}",
-                DapBatchBucket::FixedSize {
+                DapBatchBucket::LeaderSelected {
                     batch_id: BatchId([34; 32]),
                     shard: 2323,
                 }
diff --git a/crates/daphne/src/hpke.rs b/crates/daphne/src/hpke.rs
index a791781..dd8cd5e 100644
--- a/crates/daphne/src/hpke.rs
+++ b/crates/daphne/src/hpke.rs
@@ -629,7 +629,7 @@ mod test {
             sender: DapAggregatorRole::Leader,
             task_id: &crate::messages::TaskId(rand::random()),
             agg_param: &crate::DapAggregationParam::Empty,
-            batch_selector: &crate::messages::BatchSelector::FixedSizeByBatchId {
+            batch_selector: &crate::messages::BatchSelector::LeaderSelectedByBatchId {
                 batch_id: BatchId(rand::random()),
             },
         };
@@ -696,7 +696,7 @@ mod test {
                     duration: rand::random(),
                 },
             },
-            BatchSelector::FixedSizeByBatchId {
+            BatchSelector::LeaderSelectedByBatchId {
                 batch_id: BatchId(rand::random()),
             },
         ];
diff --git a/crates/daphne/src/lib.rs b/crates/daphne/src/lib.rs
index 45b4d74..4330a66 100644
--- a/crates/daphne/src/lib.rs
+++ b/crates/daphne/src/lib.rs
@@ -232,10 +232,10 @@ pub enum DapQueryConfig {
     /// specified by the query.
     TimeInterval,
 
-    /// The "fixed-size" query type where by the Leader assigns reports to arbitrary batches
-    /// identified by batch IDs. This type includes an optional maximum batch size: if set, then
+    /// The "leader-selected" query type where by the Leader assigns reports to arbitrary batches
+    /// identified by batch IDs. In draft-09 this type includes an optional maximum batch size: if set, then
     /// Aggregators are meant to stop aggregating reports when this limit is reached.
-    FixedSize {
+    LeaderSelected {
         #[serde(default)]
         max_batch_size: Option<NonZeroU32>,
     },
@@ -249,8 +249,8 @@ impl DapQueryConfig {
                 Self::TimeInterval { .. },
                 PartialBatchSelector::TimeInterval
             ) | (
-                Self::FixedSize { .. },
-                PartialBatchSelector::FixedSizeByBatchId { .. }
+                Self::LeaderSelected { .. },
+                PartialBatchSelector::LeaderSelectedByBatchId { .. }
             )
         )
     }
@@ -262,8 +262,8 @@ impl DapQueryConfig {
                 Self::TimeInterval { .. },
                 BatchSelector::TimeInterval { .. }
             ) | (
-                Self::FixedSize { .. },
-                BatchSelector::FixedSizeByBatchId { .. }
+                Self::LeaderSelected { .. },
+                BatchSelector::LeaderSelectedByBatchId { .. }
             )
         )
     }
@@ -273,7 +273,7 @@ impl std::fmt::Display for DapQueryConfig {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::TimeInterval => write!(f, "time_interval"),
-            Self::FixedSize { .. } => write!(f, "fixed_size"),
+            Self::LeaderSelected { .. } => write!(f, "leader_selected"),
         }
     }
 }
@@ -282,12 +282,12 @@ impl std::fmt::Display for DapQueryConfig {
 ///
 /// A bucket is the smallest, disjoint set of reports that can be queried: For time-interval
 /// queries, the bucket to which a report is assigned is determined by truncating its timestamp by
-/// the task's `time_precision` parameter; for fixed-size queries, the span consists of a single
+/// the task's `time_precision` parameter; for `leader_selected` queries, the span consists of a single
 /// bucket, which is the batch determined by the batch ID (i.e., the partial batch selector).
 #[derive(Debug, Clone, Eq, Hash, PartialEq)]
 #[cfg_attr(any(test, feature = "test-utils"), derive(deepsize::DeepSizeOf))]
 pub enum DapBatchBucket {
-    FixedSize { batch_id: BatchId, shard: usize },
+    LeaderSelected { batch_id: BatchId, shard: usize },
     TimeInterval { batch_window: Time, shard: usize },
 }
 
@@ -298,7 +298,7 @@ impl DapBatchBucket {
                 batch_window: _,
                 shard,
             }
-            | Self::FixedSize { batch_id: _, shard } => *shard,
+            | Self::LeaderSelected { batch_id: _, shard } => *shard,
         }
     }
 }
@@ -307,7 +307,7 @@ impl std::fmt::Display for DapBatchBucket {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::TimeInterval { batch_window, .. } => write!(f, "window/{batch_window}")?,
-            Self::FixedSize { batch_id, .. } => write!(f, "batch/{batch_id}")?,
+            Self::LeaderSelected { batch_id, .. } => write!(f, "batch/{batch_id}")?,
         };
 
         let shard = self.shard();
@@ -389,10 +389,12 @@ impl DapAggregateSpan<DapAggregateShare> {
                 batch_window: task_config.quantized_time_lower_bound(time),
                 shard,
             },
-            PartialBatchSelector::FixedSizeByBatchId { batch_id } => DapBatchBucket::FixedSize {
-                batch_id: *batch_id,
-                shard,
-            },
+            PartialBatchSelector::LeaderSelectedByBatchId { batch_id } => {
+                DapBatchBucket::LeaderSelected {
+                    batch_id: *batch_id,
+                    shard,
+                }
+            }
         };
 
         let (agg_share, reports) = self.span.entry(bucket).or_default();
@@ -707,10 +709,10 @@ impl DapTaskConfig {
                 }
                 Ok(span)
             }
-            BatchSelector::FixedSizeByBatchId { batch_id } => {
+            BatchSelector::LeaderSelectedByBatchId { batch_id } => {
                 let mut span = HashSet::with_capacity(num_agg_span_shards);
                 for shard in 0..num_agg_span_shards {
-                    span.insert(DapBatchBucket::FixedSize {
+                    span.insert(DapBatchBucket::LeaderSelected {
                         batch_id: *batch_id,
                         shard,
                     });
@@ -727,7 +729,7 @@ impl DapTaskConfig {
         report_count: u64,
     ) -> Result<bool, DapAbort> {
         match self.query {
-            DapQueryConfig::FixedSize {
+            DapQueryConfig::LeaderSelected {
                 max_batch_size: Some(max_batch_size),
             } => {
                 if report_count > u64::from(max_batch_size.get()) {
@@ -740,7 +742,7 @@ impl DapTaskConfig {
                 }
             }
             DapQueryConfig::TimeInterval
-            | DapQueryConfig::FixedSize {
+            | DapQueryConfig::LeaderSelected {
                 max_batch_size: None,
             } => (),
         };
diff --git a/crates/daphne/src/messages/mod.rs b/crates/daphne/src/messages/mod.rs
index d776d02..28d674b 100644
--- a/crates/daphne/src/messages/mod.rs
+++ b/crates/daphne/src/messages/mod.rs
@@ -25,11 +25,13 @@ use std::{
 
 // Query types
 const QUERY_TYPE_TIME_INTERVAL: u8 = 0x01;
-const QUERY_TYPE_FIXED_SIZE: u8 = 0x02;
+const QUERY_TYPE_LEADER_SELECTED: u8 = 0x02;
 
-// FixedSize query subtypes
-const FIXED_SIZE_QUERY_TYPE_BY_BATCH_ID: u8 = 0x00;
-const FIXED_SIZE_QUERY_TYPE_CURRENT_BATCH: u8 = 0x01;
+// LeaderSelected query subtypes
+// This is a slight misnomer, as these code points only exist in draft-09
+// where "leader selected" queries were still called fixed size.
+const LEADER_SELECTED_QUERY_TYPE_BY_BATCH_ID: u8 = 0x00;
+const LEADER_SELECTED_QUERY_TYPE_CURRENT_BATCH: u8 = 0x01;
 
 // Known extension types.
 const EXTENSION_TASKPROV: u16 = 0xff00;
@@ -348,14 +350,14 @@ impl ParameterizedDecode<DapVersion> for ReportShare {
 #[cfg_attr(any(test, feature = "test-utils"), derive(deepsize::DeepSizeOf))]
 pub enum PartialBatchSelector {
     TimeInterval,
-    FixedSizeByBatchId { batch_id: BatchId },
+    LeaderSelectedByBatchId { batch_id: BatchId },
 }
 
 impl std::fmt::Display for PartialBatchSelector {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::TimeInterval => write!(f, "time_interval"),
-            Self::FixedSizeByBatchId { .. } => write!(f, "fixed_size"),
+            Self::LeaderSelectedByBatchId { .. } => write!(f, "leader_selected"),
         }
     }
 }
@@ -364,7 +366,9 @@ impl From<BatchSelector> for PartialBatchSelector {
     fn from(batch_sel: BatchSelector) -> Self {
         match batch_sel {
             BatchSelector::TimeInterval { .. } => Self::TimeInterval,
-            BatchSelector::FixedSizeByBatchId { batch_id } => Self::FixedSizeByBatchId { batch_id },
+            BatchSelector::LeaderSelectedByBatchId { batch_id } => {
+                Self::LeaderSelectedByBatchId { batch_id }
+            }
         }
     }
 }
@@ -379,8 +383,8 @@ impl ParameterizedEncode<DapVersion> for PartialBatchSelector {
             DapVersion::Draft09 => {
                 match self {
                     Self::TimeInterval => QUERY_TYPE_TIME_INTERVAL.encode(bytes)?,
-                    Self::FixedSizeByBatchId { batch_id } => {
-                        QUERY_TYPE_FIXED_SIZE.encode(bytes)?;
+                    Self::LeaderSelectedByBatchId { batch_id } => {
+                        QUERY_TYPE_LEADER_SELECTED.encode(bytes)?;
                         batch_id.encode(bytes)?;
                     }
                 };
@@ -392,8 +396,8 @@ impl ParameterizedEncode<DapVersion> for PartialBatchSelector {
                         QUERY_TYPE_TIME_INTERVAL.encode(bytes)?;
                         encode_u16_bytes(bytes, &Vec::new())?;
                     }
-                    Self::FixedSizeByBatchId { batch_id } => {
-                        QUERY_TYPE_FIXED_SIZE.encode(bytes)?;
+                    Self::LeaderSelectedByBatchId { batch_id } => {
+                        QUERY_TYPE_LEADER_SELECTED.encode(bytes)?;
                         let config = &mut Vec::new();
                         batch_id.encode(config)?;
                         encode_u16_bytes(bytes, config)?;
@@ -413,7 +417,7 @@ impl ParameterizedDecode<DapVersion> for PartialBatchSelector {
         match version {
             DapVersion::Draft09 => match u8::decode(bytes)? {
                 QUERY_TYPE_TIME_INTERVAL => Ok(Self::TimeInterval),
-                QUERY_TYPE_FIXED_SIZE => Ok(Self::FixedSizeByBatchId {
+                QUERY_TYPE_LEADER_SELECTED => Ok(Self::LeaderSelectedByBatchId {
                     batch_id: BatchId::decode(bytes)?,
                 }),
                 _ => Err(CodecError::UnexpectedValue),
@@ -427,10 +431,10 @@ impl ParameterizedDecode<DapVersion> for PartialBatchSelector {
                         Err(CodecError::UnexpectedValue)
                     }
                 }
-                QUERY_TYPE_FIXED_SIZE => {
+                QUERY_TYPE_LEADER_SELECTED => {
                     let config = decode_u16_bytes(bytes)?;
                     let batch_id = BatchId::decode(&mut Cursor::new(config.as_slice()))?;
-                    Ok(Self::FixedSizeByBatchId { batch_id })
+                    Ok(Self::LeaderSelectedByBatchId { batch_id })
                 }
                 _ => Err(CodecError::UnexpectedValue),
             },
@@ -444,15 +448,19 @@ impl ParameterizedDecode<DapVersion> for PartialBatchSelector {
 #[cfg_attr(any(test, feature = "test-utils"), derive(deepsize::DeepSizeOf))]
 pub enum BatchSelector {
     TimeInterval { batch_interval: Interval },
-    FixedSizeByBatchId { batch_id: BatchId },
+    LeaderSelectedByBatchId { batch_id: BatchId },
 }
 
 impl std::fmt::Display for BatchSelector {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::TimeInterval { .. } => write!(f, "time_interval"),
-            Self::FixedSizeByBatchId { batch_id } => {
-                write!(f, "fixed_size_by_batch_id({})", batch_id.to_base64url())
+            Self::LeaderSelectedByBatchId { batch_id } => {
+                write!(
+                    f,
+                    "leader_selected_by_batch_id({})",
+                    batch_id.to_base64url()
+                )
             }
         }
     }
@@ -471,8 +479,8 @@ impl ParameterizedEncode<DapVersion> for BatchSelector {
                         QUERY_TYPE_TIME_INTERVAL.encode(bytes)?;
                         batch_interval.encode(bytes)?;
                     }
-                    Self::FixedSizeByBatchId { batch_id } => {
-                        QUERY_TYPE_FIXED_SIZE.encode(bytes)?;
+                    Self::LeaderSelectedByBatchId { batch_id } => {
+                        QUERY_TYPE_LEADER_SELECTED.encode(bytes)?;
                         batch_id.encode(bytes)?;
                     }
                 };
@@ -486,8 +494,8 @@ impl ParameterizedEncode<DapVersion> for BatchSelector {
                         batch_interval.encode(config)?;
                         encode_u16_bytes(bytes, config)?;
                     }
-                    Self::FixedSizeByBatchId { batch_id } => {
-                        QUERY_TYPE_FIXED_SIZE.encode(bytes)?;
+                    Self::LeaderSelectedByBatchId { batch_id } => {
+                        QUERY_TYPE_LEADER_SELECTED.encode(bytes)?;
                         let config = &mut Vec::new();
                         batch_id.encode(config)?;
                         encode_u16_bytes(bytes, config)?;
@@ -509,7 +517,7 @@ impl ParameterizedDecode<DapVersion> for BatchSelector {
                 QUERY_TYPE_TIME_INTERVAL => Ok(Self::TimeInterval {
                     batch_interval: Interval::decode(bytes)?,
                 }),
-                QUERY_TYPE_FIXED_SIZE => Ok(Self::FixedSizeByBatchId {
+                QUERY_TYPE_LEADER_SELECTED => Ok(Self::LeaderSelectedByBatchId {
                     batch_id: BatchId::decode(bytes)?,
                 }),
                 _ => Err(CodecError::UnexpectedValue),
@@ -521,9 +529,9 @@ impl ParameterizedDecode<DapVersion> for BatchSelector {
                         batch_interval: Interval::decode(&mut Cursor::new(config.as_slice()))?,
                     })
                 }
-                QUERY_TYPE_FIXED_SIZE => {
+                QUERY_TYPE_LEADER_SELECTED => {
                     let config = decode_u16_bytes(bytes)?;
-                    Ok(Self::FixedSizeByBatchId {
+                    Ok(Self::LeaderSelectedByBatchId {
                         batch_id: BatchId::decode(&mut Cursor::new(config.as_slice()))?,
                     })
                 }
@@ -1001,8 +1009,8 @@ impl Decode for Interval {
 #[cfg_attr(any(test, feature = "test-utils"), derive(deepsize::DeepSizeOf))]
 pub enum Query {
     TimeInterval { batch_interval: Interval },
-    FixedSizeByBatchId { batch_id: BatchId },
-    FixedSizeCurrentBatch,
+    LeaderSelectedByBatchId { batch_id: BatchId },
+    LeaderSelectedCurrentBatch,
 }
 
 impl Query {
@@ -1011,10 +1019,10 @@ impl Query {
             Self::TimeInterval { batch_interval } => {
                 Some(BatchSelector::TimeInterval { batch_interval })
             }
-            Self::FixedSizeByBatchId { batch_id } => {
-                Some(BatchSelector::FixedSizeByBatchId { batch_id })
+            Self::LeaderSelectedByBatchId { batch_id } => {
+                Some(BatchSelector::LeaderSelectedByBatchId { batch_id })
             }
-            Self::FixedSizeCurrentBatch => None,
+            Self::LeaderSelectedCurrentBatch => None,
         }
     }
 }
@@ -1023,7 +1031,9 @@ impl From<BatchSelector> for Query {
     fn from(batch_sel: BatchSelector) -> Self {
         match batch_sel {
             BatchSelector::TimeInterval { batch_interval } => Self::TimeInterval { batch_interval },
-            BatchSelector::FixedSizeByBatchId { batch_id } => Self::FixedSizeByBatchId { batch_id },
+            BatchSelector::LeaderSelectedByBatchId { batch_id } => {
+                Self::LeaderSelectedByBatchId { batch_id }
+            }
         }
     }
 }
@@ -1032,10 +1042,14 @@ impl std::fmt::Display for Query {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::TimeInterval { .. } => write!(f, "time_interval"),
-            Self::FixedSizeByBatchId { batch_id } => {
-                write!(f, "fixed_size_by_batch_id({})", batch_id.to_base64url())
+            Self::LeaderSelectedByBatchId { batch_id } => {
+                write!(
+                    f,
+                    "leader_selected_by_batch_id({})",
+                    batch_id.to_base64url()
+                )
             }
-            Self::FixedSizeCurrentBatch => write!(f, "fixed_size_current_batch"),
+            Self::LeaderSelectedCurrentBatch => write!(f, "leader_selected_current_batch"),
         }
     }
 }
@@ -1053,14 +1067,14 @@ impl ParameterizedEncode<DapVersion> for Query {
                         QUERY_TYPE_TIME_INTERVAL.encode(bytes)?;
                         batch_interval.encode(bytes)?;
                     }
-                    Self::FixedSizeByBatchId { batch_id } => {
-                        QUERY_TYPE_FIXED_SIZE.encode(bytes)?;
-                        FIXED_SIZE_QUERY_TYPE_BY_BATCH_ID.encode(bytes)?;
+                    Self::LeaderSelectedByBatchId { batch_id } => {
+                        QUERY_TYPE_LEADER_SELECTED.encode(bytes)?;
+                        LEADER_SELECTED_QUERY_TYPE_BY_BATCH_ID.encode(bytes)?;
                         batch_id.encode(bytes)?;
                     }
-                    Self::FixedSizeCurrentBatch => {
-                        QUERY_TYPE_FIXED_SIZE.encode(bytes)?;
-                        FIXED_SIZE_QUERY_TYPE_CURRENT_BATCH.encode(bytes)?;
+                    Self::LeaderSelectedCurrentBatch => {
+                        QUERY_TYPE_LEADER_SELECTED.encode(bytes)?;
+                        LEADER_SELECTED_QUERY_TYPE_CURRENT_BATCH.encode(bytes)?;
                     }
                 };
             }
@@ -1071,11 +1085,11 @@ impl ParameterizedEncode<DapVersion> for Query {
                     batch_interval.encode(config)?;
                     encode_u16_bytes(bytes, config)?;
                 }
-                Self::FixedSizeByBatchId { batch_id: _ } => {
+                Self::LeaderSelectedByBatchId { batch_id: _ } => {
                     return Err(CodecError::UnexpectedValue)
                 }
-                Self::FixedSizeCurrentBatch => {
-                    QUERY_TYPE_FIXED_SIZE.encode(bytes)?;
+                Self::LeaderSelectedCurrentBatch => {
+                    QUERY_TYPE_LEADER_SELECTED.encode(bytes)?;
                     let config = &mut Vec::new();
                     encode_u16_bytes(bytes, config)?;
                 }
@@ -1095,13 +1109,17 @@ impl ParameterizedDecode<DapVersion> for Query {
                 QUERY_TYPE_TIME_INTERVAL => Ok(Self::TimeInterval {
                     batch_interval: Interval::decode(bytes)?,
                 }),
-                QUERY_TYPE_FIXED_SIZE => {
+                QUERY_TYPE_LEADER_SELECTED => {
                     let subtype = u8::decode(bytes)?;
                     match subtype {
-                        FIXED_SIZE_QUERY_TYPE_BY_BATCH_ID => Ok(Self::FixedSizeByBatchId {
-                            batch_id: BatchId::decode(bytes)?,
-                        }),
-                        FIXED_SIZE_QUERY_TYPE_CURRENT_BATCH => Ok(Self::FixedSizeCurrentBatch),
+                        LEADER_SELECTED_QUERY_TYPE_BY_BATCH_ID => {
+                            Ok(Self::LeaderSelectedByBatchId {
+                                batch_id: BatchId::decode(bytes)?,
+                            })
+                        }
+                        LEADER_SELECTED_QUERY_TYPE_CURRENT_BATCH => {
+                            Ok(Self::LeaderSelectedCurrentBatch)
+                        }
                         _ => Err(CodecError::UnexpectedValue),
                     }
                 }
@@ -1114,12 +1132,12 @@ impl ParameterizedDecode<DapVersion> for Query {
                         batch_interval: Interval::decode(&mut Cursor::new(config.as_slice()))?,
                     })
                 }
-                QUERY_TYPE_FIXED_SIZE => {
+                QUERY_TYPE_LEADER_SELECTED => {
                     let config = decode_u16_bytes(bytes)?;
                     if !config.is_empty() {
                         Err(CodecError::UnexpectedValue)
                     } else {
-                        Ok(Self::FixedSizeCurrentBatch)
+                        Ok(Self::LeaderSelectedCurrentBatch)
                     }
                 }
                 _ => Err(CodecError::UnexpectedValue),
@@ -1685,7 +1703,7 @@ mod test {
 
         let want = AggregationJobInitReq {
             agg_param: b"this is an aggregation parameter".to_vec(),
-            part_batch_sel: PartialBatchSelector::FixedSizeByBatchId {
+            part_batch_sel: PartialBatchSelector::LeaderSelectedByBatchId {
                 batch_id: BatchId([0; 32]),
             },
             prep_inits: vec![
@@ -1739,7 +1757,7 @@ mod test {
     fn roundtrip_agg_job_init_req(version: DapVersion) {
         let want = AggregationJobInitReq {
             agg_param: b"this is an aggregation parameter".to_vec(),
-            part_batch_sel: PartialBatchSelector::FixedSizeByBatchId {
+            part_batch_sel: PartialBatchSelector::LeaderSelectedByBatchId {
                 batch_id: BatchId([0; 32]),
             },
             prep_inits: vec![
@@ -1830,7 +1848,7 @@ mod test {
     #[test]
     fn read_agg_share_req() {
         let want = AggregateShareReq {
-            batch_sel: BatchSelector::FixedSizeByBatchId {
+            batch_sel: BatchSelector::LeaderSelectedByBatchId {
                 batch_id: BatchId([23; 32]),
             },
             agg_param: b"this is an aggregation parameter".to_vec(),
diff --git a/crates/daphne/src/messages/taskprov.rs b/crates/daphne/src/messages/taskprov.rs
index d833cbe..1824b2b 100644
--- a/crates/daphne/src/messages/taskprov.rs
+++ b/crates/daphne/src/messages/taskprov.rs
@@ -4,7 +4,7 @@
 //! draft-wang-ppm-dap-taskprov: Messages for the taskrpov extension for DAP.
 
 use crate::messages::{
-    decode_u16_bytes, encode_u16_bytes, Duration, Time, QUERY_TYPE_FIXED_SIZE,
+    decode_u16_bytes, encode_u16_bytes, Duration, Time, QUERY_TYPE_LEADER_SELECTED,
     QUERY_TYPE_TIME_INTERVAL,
 };
 use crate::pine::PineParam;
@@ -319,7 +319,7 @@ impl Decode for UrlBytes {
 #[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
 pub enum QueryConfigVar {
     TimeInterval,
-    FixedSize { max_batch_size: Option<NonZeroU32> },
+    LeaderSelected { max_batch_size: Option<NonZeroU32> },
     NotImplemented { typ: u8, param: Vec<u8> },
 }
 
@@ -338,8 +338,8 @@ impl QueryConfig {
             QueryConfigVar::TimeInterval => {
                 QUERY_TYPE_TIME_INTERVAL.encode(bytes)?;
             }
-            QueryConfigVar::FixedSize { .. } => {
-                QUERY_TYPE_FIXED_SIZE.encode(bytes)?;
+            QueryConfigVar::LeaderSelected { .. } => {
+                QUERY_TYPE_LEADER_SELECTED.encode(bytes)?;
             }
             QueryConfigVar::NotImplemented { typ, .. } => {
                 typ.encode(bytes)?;
@@ -363,7 +363,7 @@ impl ParameterizedEncode<DapVersion> for QueryConfig {
         self.encode_query_type(bytes)?;
         match &self.var {
             QueryConfigVar::TimeInterval => (),
-            QueryConfigVar::FixedSize { max_batch_size } => {
+            QueryConfigVar::LeaderSelected { max_batch_size } => {
                 match version {
                     DapVersion::Draft09 => match max_batch_size {
                         Some(x) => x.get().encode(bytes)?,
@@ -399,7 +399,7 @@ impl ParameterizedDecode<(DapVersion, Option<usize>)> for QueryConfig {
         let var =
             match (bytes_left, query_type) {
                 (.., QUERY_TYPE_TIME_INTERVAL) => QueryConfigVar::TimeInterval,
-                (.., QUERY_TYPE_FIXED_SIZE) => QueryConfigVar::FixedSize {
+                (.., QUERY_TYPE_LEADER_SELECTED) => QueryConfigVar::LeaderSelected {
                     max_batch_size: match version {
                         DapVersion::Draft09 => NonZeroU32::new(u32::decode(bytes)?),
                         DapVersion::Latest => None,
@@ -559,7 +559,7 @@ mod tests {
                     DapVersion::Latest => 1,
                 },
                 min_batch_size: 55,
-                var: QueryConfigVar::FixedSize {
+                var: QueryConfigVar::LeaderSelected {
                     max_batch_size: match version {
                         DapVersion::Draft09 => Some(NonZeroU32::new(57).unwrap()),
                         DapVersion::Latest => None,
@@ -627,7 +627,7 @@ mod tests {
                 DapVersion::Latest => 1,
             },
             min_batch_size: 12_345_678,
-            var: QueryConfigVar::FixedSize {
+            var: QueryConfigVar::LeaderSelected {
                 max_batch_size: match version {
                     DapVersion::Draft09 => Some(NonZeroU32::new(12_345_678).unwrap()),
                     DapVersion::Latest => None,
@@ -801,7 +801,7 @@ mod tests {
                     DapVersion::Latest => 1,
                 },
                 min_batch_size: 55,
-                var: QueryConfigVar::FixedSize {
+                var: QueryConfigVar::LeaderSelected {
                     max_batch_size: match version {
                         DapVersion::Draft09 => Some(NonZeroU32::new(57).unwrap()),
                         DapVersion::Latest => None,
diff --git a/crates/daphne/src/roles/aggregator.rs b/crates/daphne/src/roles/aggregator.rs
index 2bb3997..0d1efcf 100644
--- a/crates/daphne/src/roles/aggregator.rs
+++ b/crates/daphne/src/roles/aggregator.rs
@@ -85,7 +85,7 @@ pub trait DapAggregator: HpkeProvider + Sized {
     ) -> Result<bool, DapError>;
 
     /// Check whether the given batch ID has been observed before. This is called by the Leader
-    /// (resp. Helper) in response to a CollectReq (resp. AggregateShareReq) for fixed-size tasks.
+    /// (resp. Helper) in response to a CollectReq (resp. AggregateShareReq) for leader-selected tasks.
     async fn batch_exists(&self, task_id: &TaskId, batch_id: &BatchId) -> Result<bool, DapError>;
 
     /// Store a set of output shares and mark the corresponding reports as aggregated.
diff --git a/crates/daphne/src/roles/leader/in_memory_leader.rs b/crates/daphne/src/roles/leader/in_memory_leader.rs
index aa66636..3d34aee 100644
--- a/crates/daphne/src/roles/leader/in_memory_leader.rs
+++ b/crates/daphne/src/roles/leader/in_memory_leader.rs
@@ -81,9 +81,9 @@ impl InMemoryLeaderState {
         task_id: &TaskId,
         task_config: &DapTaskConfig,
     ) -> std::result::Result<BatchId, DapError> {
-        if !matches!(task_config.query, DapQueryConfig::FixedSize { .. }) {
+        if !matches!(task_config.query, DapQueryConfig::LeaderSelected { .. }) {
             return Err(DapError::Abort(DapAbort::BadRequest(
-                "tried to get current batch from non fixed-size task".into(),
+                "tried to get current batch from non leader-selected task".into(),
             )));
         }
 
@@ -151,7 +151,7 @@ impl InMemoryLeaderState {
             }
 
             // The batch will be collected, so remove it from the batch queue.
-            if let DapBatchBucket::FixedSize {
+            if let DapBatchBucket::LeaderSelected {
                 ref batch_id,
                 shard: _,
             } = bucket
@@ -249,13 +249,13 @@ impl MockLeaderMemoryPerTask {
             .shard(NonZeroUsize::new(1).unwrap());
 
         match task_config.query {
-            // For fixed-size queries, the bucket corresponds to a single batch.
-            DapQueryConfig::FixedSize { .. } => {
+            // For leader-selected queries, the bucket corresponds to a single batch.
+            DapQueryConfig::LeaderSelected { .. } => {
                 // Assign the report to the first unsaturated batch.
                 for (batch_id, report_count) in &mut self.batch_queue {
                     if *report_count < task_config.min_batch_size {
                         *report_count += 1;
-                        return DapBatchBucket::FixedSize {
+                        return DapBatchBucket::LeaderSelected {
                             batch_id: *batch_id,
                             shard,
                         };
@@ -265,7 +265,7 @@ impl MockLeaderMemoryPerTask {
                 // No unsaturated batch exists, so create a new batch.
                 let batch_id = BatchId(rng.gen());
                 self.batch_queue.push_back((batch_id, 1));
-                DapBatchBucket::FixedSize { batch_id, shard }
+                DapBatchBucket::LeaderSelected { batch_id, shard }
             }
 
             // For time-interval queries, the bucket is the batch window computed by truncating the
diff --git a/crates/daphne/src/roles/leader/mod.rs b/crates/daphne/src/roles/leader/mod.rs
index c0076cb..5034618 100644
--- a/crates/daphne/src/roles/leader/mod.rs
+++ b/crates/daphne/src/roles/leader/mod.rs
@@ -114,7 +114,7 @@ pub trait DapLeader: DapAggregator {
     /// Store a report for use later on.
     async fn put_report(&self, report: &Report, task_id: &TaskId) -> Result<(), DapError>;
 
-    /// Fixed-size tasks: Return the ID of the batch currently being filled.
+    /// Leader selected tasks: Return the ID of the batch currently being filled.
     //
     // TODO draft02 cleanup: Consider removing this.
     async fn current_batch(&self, task_id: &TaskId) -> Result<BatchId, DapError>;
@@ -269,8 +269,10 @@ pub async fn handle_coll_job_req<A: DapLeader>(
 
     let batch_sel = match coll_job_req.query {
         Query::TimeInterval { batch_interval } => BatchSelector::TimeInterval { batch_interval },
-        Query::FixedSizeByBatchId { batch_id } => BatchSelector::FixedSizeByBatchId { batch_id },
-        Query::FixedSizeCurrentBatch => BatchSelector::FixedSizeByBatchId {
+        Query::LeaderSelectedByBatchId { batch_id } => {
+            BatchSelector::LeaderSelectedByBatchId { batch_id }
+        }
+        Query::LeaderSelectedCurrentBatch => BatchSelector::LeaderSelectedByBatchId {
             batch_id: aggregator.current_batch(&task_id).await?,
         },
     };
diff --git a/crates/daphne/src/roles/mod.rs b/crates/daphne/src/roles/mod.rs
index 15153d1..da0b8b2 100644
--- a/crates/daphne/src/roles/mod.rs
+++ b/crates/daphne/src/roles/mod.rs
@@ -65,8 +65,8 @@ async fn check_batch(
                 );
             }
         }
-        (DapQueryConfig::FixedSize { .. }, Query::FixedSizeCurrentBatch) => (), // nothing to do
-        (DapQueryConfig::FixedSize { .. }, Query::FixedSizeByBatchId { batch_id }) => {
+        (DapQueryConfig::LeaderSelected { .. }, Query::LeaderSelectedCurrentBatch) => (), // nothing to do
+        (DapQueryConfig::LeaderSelected { .. }, Query::LeaderSelectedByBatchId { batch_id }) => {
             if !agg.batch_exists(task_id, batch_id).await? {
                 return Err(DapAbort::BatchInvalid {
                     detail: format!(
@@ -166,7 +166,7 @@ mod test {
         pub now: Time,
         global_config: DapGlobalConfig,
         pub time_interval_task_id: TaskId,
-        pub fixed_size_task_id: TaskId,
+        pub leader_selected_task_id: TaskId,
         pub expired_task_id: TaskId,
         #[cfg(feature = "experimental")]
         pub mastic_task_id: TaskId,
@@ -206,7 +206,7 @@ mod test {
 
             // Create the task list.
             let time_interval_task_id = TaskId(rng.gen());
-            let fixed_size_task_id = TaskId(rng.gen());
+            let leader_selected_task_id = TaskId(rng.gen());
             let expired_task_id = TaskId(rng.gen());
             #[cfg(feature = "experimental")]
             let mastic_task_id = TaskId(rng.gen());
@@ -230,7 +230,7 @@ mod test {
                 },
             );
             tasks.insert(
-                fixed_size_task_id,
+                leader_selected_task_id,
                 DapTaskConfig {
                     version,
                     collector_hpke_config: collector_hpke_receiver_config.config.clone(),
@@ -240,7 +240,7 @@ mod test {
                     not_before: now,
                     not_after: now + Self::TASK_TIME_PRECISION,
                     min_batch_size: 1,
-                    query: DapQueryConfig::FixedSize {
+                    query: DapQueryConfig::LeaderSelected {
                         max_batch_size: Some(NonZeroU32::new(2).unwrap()),
                     },
                     vdaf: vdaf_config,
@@ -318,7 +318,7 @@ mod test {
                 now,
                 global_config,
                 time_interval_task_id,
-                fixed_size_task_id,
+                leader_selected_task_id,
                 expired_task_id,
                 #[cfg(feature = "experimental")]
                 mastic_task_id,
@@ -361,7 +361,7 @@ mod test {
                 leader,
                 helper,
                 time_interval_task_id: self.time_interval_task_id,
-                fixed_size_task_id: self.fixed_size_task_id,
+                leader_selected_task_id: self.leader_selected_task_id,
                 expired_task_id: self.expired_task_id,
                 #[cfg(feature = "experimental")]
                 mastic_task_id: self.mastic_task_id,
@@ -376,7 +376,7 @@ mod test {
         leader: Arc<InMemoryAggregator>,
         helper: Arc<InMemoryAggregator>,
         time_interval_task_id: TaskId,
-        fixed_size_task_id: TaskId,
+        leader_selected_task_id: TaskId,
         expired_task_id: TaskId,
         #[cfg(feature = "experimental")]
         mastic_task_id: TaskId,
@@ -449,9 +449,11 @@ mod test {
             let task_config = self.leader.unchecked_get_task_config(task_id).await;
             let part_batch_sel = match task_config.query {
                 DapQueryConfig::TimeInterval { .. } => PartialBatchSelector::TimeInterval,
-                DapQueryConfig::FixedSize { .. } => PartialBatchSelector::FixedSizeByBatchId {
-                    batch_id: BatchId(rng.gen()),
-                },
+                DapQueryConfig::LeaderSelected { .. } => {
+                    PartialBatchSelector::LeaderSelectedByBatchId {
+                        batch_id: BatchId(rng.gen()),
+                    }
+                }
             };
 
             let agg_job_id = AggregationJobId(rng.gen());
@@ -572,7 +574,7 @@ mod test {
         let task_config = t.leader.unchecked_get_task_config(task_id).await;
         let agg_job_id = AggregationJobId(rng.gen());
 
-        // Helper expects "time_interval" query, but Leader indicates "fixed_size".
+        // Helper expects "time_interval" query, but Leader indicates "leader_selected".
         let req = Test::leader_req(
             task_id,
             &task_config,
@@ -580,7 +582,7 @@ mod test {
             DapMediaType::AggregationJobInitReq,
             AggregationJobInitReq {
                 agg_param: Vec::default(),
-                part_batch_sel: PartialBatchSelector::FixedSizeByBatchId {
+                part_batch_sel: PartialBatchSelector::LeaderSelectedByBatchId {
                     batch_id: BatchId(rng.gen()),
                 },
                 prep_inits: Vec::default(),
@@ -662,7 +664,7 @@ mod test {
         let mut rng = thread_rng();
         let t = Test::new(version);
 
-        // Helper expects "time_interval" query, but Leader sent "fixed_size".
+        // Helper expects "time_interval" query, but Leader sent "leader_selected".
         let task_config = t
             .leader
             .unchecked_get_task_config(&t.time_interval_task_id)
@@ -673,7 +675,7 @@ mod test {
             (),
             DapMediaType::AggregateShareReq,
             AggregateShareReq {
-                batch_sel: BatchSelector::FixedSizeByBatchId {
+                batch_sel: BatchSelector::LeaderSelectedByBatchId {
                     batch_id: BatchId(rng.gen()),
                 },
                 agg_param: Vec::default(),
@@ -691,15 +693,15 @@ mod test {
         // Leader sends aggregate share request for unrecognized batch ID.
         let task_config = t
             .leader
-            .unchecked_get_task_config(&t.fixed_size_task_id)
+            .unchecked_get_task_config(&t.leader_selected_task_id)
             .await;
         let req = Test::leader_req(
-            &t.fixed_size_task_id,
+            &t.leader_selected_task_id,
             &task_config,
             (),
             DapMediaType::AggregateShareReq,
             AggregateShareReq {
-                batch_sel: BatchSelector::FixedSizeByBatchId {
+                batch_sel: BatchSelector::LeaderSelectedByBatchId {
                     batch_id: BatchId(rng.gen()), // Unrecognized batch ID
                 },
                 agg_param: Vec::default(),
@@ -1204,11 +1206,11 @@ mod test {
 
     async fn handle_coll_job_req_fail_unrecongized_batch(version: DapVersion) {
         let t = Test::new(version);
-        let task_id = &t.fixed_size_task_id;
+        let task_id = &t.leader_selected_task_id;
 
         let req = t
             .gen_test_coll_job_req(
-                Query::FixedSizeByBatchId {
+                Query::LeaderSelectedByBatchId {
                     batch_id: BatchId(thread_rng().gen()),
                 },
                 task_id,
@@ -1272,7 +1274,7 @@ mod test {
         let mut rng = thread_rng();
         let t = Test::new(version);
 
-        // Leader expects "time_interval" query, but Collector sent "fixed_size".
+        // Leader expects "time_interval" query, but Collector sent "leader_selected".
         let task_config = t
             .leader
             .unchecked_get_task_config(&t.time_interval_task_id)
@@ -1282,7 +1284,7 @@ mod test {
             &task_config,
             DapMediaType::CollectionReq,
             CollectionReq {
-                query: Query::FixedSizeByBatchId {
+                query: Query::LeaderSelectedByBatchId {
                     batch_id: BatchId(rng.gen()),
                 },
                 agg_param: Vec::default(),
@@ -1298,14 +1300,14 @@ mod test {
         // Collector indicates unrecognized batch ID.
         let task_config = t
             .leader
-            .unchecked_get_task_config(&t.fixed_size_task_id)
+            .unchecked_get_task_config(&t.leader_selected_task_id)
             .await;
         let req = Test::collector_req(
-            &t.fixed_size_task_id,
+            &t.leader_selected_task_id,
             &task_config,
             DapMediaType::CollectionReq,
             CollectionReq {
-                query: Query::FixedSizeByBatchId {
+                query: Query::LeaderSelectedByBatchId {
                     batch_id: BatchId(rng.gen()), // Unrecognized batch ID
                 },
                 agg_param: Vec::default(),
@@ -1340,7 +1342,7 @@ mod test {
         let task_id = &t.time_interval_task_id;
         let task_config = t
             .leader
-            .unchecked_get_task_config(&t.fixed_size_task_id)
+            .unchecked_get_task_config(&t.leader_selected_task_id)
             .await;
 
         // Client: Send upload request to Leader.
@@ -1375,9 +1377,9 @@ mod test {
 
     async_test_versions! { e2e_time_interval }
 
-    async fn e2e_fixed_size(version: DapVersion) {
+    async fn e2e_leader_selected(version: DapVersion) {
         let t = Test::new(version);
-        let task_id = &t.fixed_size_task_id;
+        let task_id = &t.leader_selected_task_id;
 
         // Client: Send upload request to Leader.
         let report = t.gen_test_report(task_id).await;
@@ -1386,7 +1388,7 @@ mod test {
             .unwrap();
 
         // Collector: Request result from the Leader.
-        let query = Query::FixedSizeCurrentBatch;
+        let query = Query::LeaderSelectedCurrentBatch;
         leader::handle_coll_job_req(&*t.leader, &t.gen_test_coll_job_req(query, task_id).await)
             .await
             .unwrap();
@@ -1409,7 +1411,7 @@ mod test {
         });
     }
 
-    async_test_versions! { e2e_fixed_size }
+    async_test_versions! { e2e_leader_selected }
 
     async fn e2e_taskprov(
         version: DapVersion,
@@ -1421,7 +1423,7 @@ mod test {
         let (task_config, task_id, taskprov_advertisement) = DapTaskParameters {
             version,
             min_batch_size: 1,
-            query: DapQueryConfig::FixedSize {
+            query: DapQueryConfig::LeaderSelected {
                 max_batch_size: Some(NonZeroU32::new(2).unwrap()),
             },
             vdaf: vdaf_config,
@@ -1474,7 +1476,7 @@ mod test {
         }
 
         // Collector: Request result from the Leader.
-        let query = Query::FixedSizeCurrentBatch;
+        let query = Query::LeaderSelectedCurrentBatch;
         leader::handle_coll_job_req(&*t.leader, &t.gen_test_coll_job_req(query, &task_id).await)
             .await
             .unwrap();
@@ -1581,7 +1583,7 @@ mod test {
             let task_id = &t.time_interval_task_id;
             let task_config = t
                 .leader
-                .unchecked_get_task_config(&t.fixed_size_task_id)
+                .unchecked_get_task_config(&t.leader_selected_task_id)
                 .await;
 
             let report = t.gen_test_report(task_id).await;
@@ -1597,7 +1599,7 @@ mod test {
 
         // Requests for the fixed-length task.
         {
-            let task_id = &t.fixed_size_task_id;
+            let task_id = &t.leader_selected_task_id;
 
             let report = t.gen_test_report(task_id).await;
             leader::handle_upload_req(&*t.leader, t.gen_test_upload_req(report, task_id).await)
@@ -1605,7 +1607,7 @@ mod test {
                 .unwrap();
 
             // Collector: Request result from the Leader.
-            let query = Query::FixedSizeCurrentBatch;
+            let query = Query::LeaderSelectedCurrentBatch;
             leader::handle_coll_job_req(&*t.leader, &t.gen_test_coll_job_req(query, task_id).await)
                 .await
                 .unwrap();
@@ -1638,7 +1640,7 @@ mod test {
         let task_id = &t.mastic_task_id;
         let task_config = t
             .leader
-            .unchecked_get_task_config(&t.fixed_size_task_id)
+            .unchecked_get_task_config(&t.leader_selected_task_id)
             .await;
 
         for i in 0..10 {
diff --git a/crates/daphne/src/taskprov.rs b/crates/daphne/src/taskprov.rs
index 5d70cf3..cfb251a 100644
--- a/crates/daphne/src/taskprov.rs
+++ b/crates/daphne/src/taskprov.rs
@@ -122,8 +122,8 @@ fn url_from_bytes(task_id: &TaskId, url_bytes: &[u8]) -> Result<Url, DapAbort> {
 impl DapQueryConfig {
     fn try_from_taskprov(task_id: &TaskId, var: QueryConfigVar) -> Result<Self, DapAbort> {
         match var {
-            QueryConfigVar::FixedSize { max_batch_size } => {
-                Ok(DapQueryConfig::FixedSize { max_batch_size })
+            QueryConfigVar::LeaderSelected { max_batch_size } => {
+                Ok(DapQueryConfig::LeaderSelected { max_batch_size })
             }
             QueryConfigVar::TimeInterval => Ok(DapQueryConfig::TimeInterval),
             QueryConfigVar::NotImplemented { typ, .. } => Err(DapAbort::InvalidTask {
@@ -354,8 +354,8 @@ impl TryFrom<&DapQueryConfig> for messages::taskprov::QueryConfigVar {
     fn try_from(query_config: &DapQueryConfig) -> Result<Self, DapError> {
         Ok(match query_config {
             DapQueryConfig::TimeInterval => messages::taskprov::QueryConfigVar::TimeInterval,
-            DapQueryConfig::FixedSize { max_batch_size } => {
-                messages::taskprov::QueryConfigVar::FixedSize {
+            DapQueryConfig::LeaderSelected { max_batch_size } => {
+                messages::taskprov::QueryConfigVar::LeaderSelected {
                     max_batch_size: *max_batch_size,
                 }
             }
@@ -472,7 +472,7 @@ mod test {
                 time_precision: 3600,
                 max_batch_query_count: 1,
                 min_batch_size: 1,
-                var: messages::taskprov::QueryConfigVar::FixedSize {
+                var: messages::taskprov::QueryConfigVar::LeaderSelected {
                     max_batch_size: Some(NonZeroU32::new(2).unwrap()),
                 },
             },
@@ -554,7 +554,7 @@ mod test {
                     time_precision: 3600,
                     max_batch_query_count: 1,
                     min_batch_size: 1,
-                    var: messages::taskprov::QueryConfigVar::FixedSize {
+                    var: messages::taskprov::QueryConfigVar::LeaderSelected {
                         max_batch_size: Some(NonZeroU32::new(2).unwrap()),
                     },
                 },
@@ -621,7 +621,7 @@ mod test {
                     time_precision: 3600,
                     max_batch_query_count: 1,
                     min_batch_size: 1,
-                    var: messages::taskprov::QueryConfigVar::FixedSize {
+                    var: messages::taskprov::QueryConfigVar::LeaderSelected {
                         max_batch_size: Some(NonZeroU32::new(2).unwrap()),
                     },
                 },


Test patch:
diff --git a/crates/dapf/src/acceptance/load_testing.rs b/crates/dapf/src/acceptance/load_testing.rs
index 2b88de2..a866d56 100644
--- a/crates/dapf/src/acceptance/load_testing.rs
+++ b/crates/dapf/src/acceptance/load_testing.rs
@@ -424,7 +424,7 @@ pub async fn execute_single_combination_from_env(
         println!("\t- load_control:        {load_control:?}");
 
         let batch_id = BatchId(thread_rng().gen());
-        let part_batch_sel = PartialBatchSelector::FixedSizeByBatchId { batch_id };
+        let part_batch_sel = PartialBatchSelector::LeaderSelectedByBatchId { batch_id };
 
         let run = async {
             let (test_task_config, hpke_config_fetch_time) = t
diff --git a/crates/daphne-server/tests/e2e/e2e.rs b/crates/daphne-server/tests/e2e/e2e.rs
index fe33ec0..17e8b99 100644
--- a/crates/daphne-server/tests/e2e/e2e.rs
+++ b/crates/daphne-server/tests/e2e/e2e.rs
@@ -1162,9 +1162,9 @@ async fn leader_collect_abort_overlapping_batch_interval(version: DapVersion) {
 async_test_versions! { leader_collect_abort_overlapping_batch_interval }
 
 #[tokio::test]
-async fn fixed_size() {
+async fn leader_selected() {
     let version = DapVersion::Draft09;
-    let t = TestRunner::fixed_size(version).await;
+    let t = TestRunner::leader_selected(version).await;
     let path = t.upload_path();
     let client = t.http_client();
     let hpke_config_list = t.get_hpke_configs(version, client).await.unwrap();
@@ -1204,7 +1204,7 @@ async fn fixed_size() {
     // Collector: Get the collect URI.
     let agg_param = DapAggregationParam::Empty;
     let collect_req = CollectionReq {
-        query: Query::FixedSizeCurrentBatch,
+        query: Query::LeaderSelectedCurrentBatch,
         agg_param: agg_param.get_encoded().unwrap(),
     };
     let collect_uri = t
@@ -1247,7 +1247,7 @@ async fn fixed_size() {
         .consume_encrypted_agg_shares(
             &t.collector_hpke_receiver,
             &t.task_id,
-            &BatchSelector::FixedSizeByBatchId { batch_id },
+            &BatchSelector::LeaderSelectedByBatchId { batch_id },
             collection.report_count,
             &agg_param,
             collection.encrypted_agg_shares.to_vec(),
@@ -1302,7 +1302,7 @@ async fn fixed_size() {
     // Collector: Get the collect URI.
     let agg_param = DapAggregationParam::Empty;
     let collect_req = CollectionReq {
-        query: Query::FixedSizeCurrentBatch,
+        query: Query::LeaderSelectedCurrentBatch,
         agg_param: agg_param.get_encoded().unwrap(),
     };
     let collect_uri = t
@@ -1329,7 +1329,7 @@ async fn fixed_size() {
         DapMediaType::CollectionReq,
         None,
         CollectionReq {
-            query: Query::FixedSizeByBatchId {
+            query: Query::LeaderSelectedByBatchId {
                 batch_id: prev_batch_id,
             },
             agg_param: Vec::new(),
diff --git a/crates/daphne-server/tests/e2e/test_runner.rs b/crates/daphne-server/tests/e2e/test_runner.rs
index b164d94..2e2f3d1 100644
--- a/crates/daphne-server/tests/e2e/test_runner.rs
+++ b/crates/daphne-server/tests/e2e/test_runner.rs
@@ -62,10 +62,10 @@ impl TestRunner {
         Self::with(version, &DapQueryConfig::TimeInterval).await
     }
 
-    pub async fn fixed_size(version: DapVersion) -> Self {
+    pub async fn leader_selected(version: DapVersion) -> Self {
         Self::with(
             version,
-            &DapQueryConfig::FixedSize {
+            &DapQueryConfig::LeaderSelected {
                 max_batch_size: Some(NonZeroU32::new(MAX_BATCH_SIZE).unwrap()),
             },
         )
@@ -176,7 +176,7 @@ impl TestRunner {
 
         let (query_type, max_batch_size) = match t.task_config.query {
             DapQueryConfig::TimeInterval => (1, None),
-            DapQueryConfig::FixedSize { max_batch_size } => (2, Some(max_batch_size)),
+            DapQueryConfig::LeaderSelected { max_batch_size } => (2, Some(max_batch_size)),
         };
 
         const MAX_ATTEMPTS: usize = 10;
diff --git a/crates/daphne/src/testing/mod.rs b/crates/daphne/src/testing/mod.rs
index c9ddb61..226686e 100644
--- a/crates/daphne/src/testing/mod.rs
+++ b/crates/daphne/src/testing/mod.rs
@@ -740,9 +740,11 @@ impl DapAggregator for InMemoryAggregator {
                 .lock()
                 .map_err(|_| fatal_error!(err = "agg_store poisoned"))?;
             let mut aggregated = false;
-            for bucket in task_config.batch_span_for_sel(&BatchSelector::FixedSizeByBatchId {
-                batch_id: *batch_id,
-            })? {
+            for bucket in
+                task_config.batch_span_for_sel(&BatchSelector::LeaderSelectedByBatchId {
+                    batch_id: *batch_id,
+                })?
+            {
                 if !agg_store.for_bucket(task_id, &bucket).agg_share.empty() {
                     aggregated = true;
                     break;