Repository: cloudflare/daphne
Pull request #120 (2022-09-14 17:10:40)

DAP-02: messages
Partially addresses #100.

Changes in this PR (each corresponds to a commit, so best to review each commit individually):
1. Bump daphne to 0.2.0 and  lock daphne_worker to daphne 0.1.2 until we're ready to upgrade to 0.2.0. This allows us to move daphne to DAP-02 without impacting daphne_worker.
2. Rework encoding of length-prefixed fields. In preparation for DAP-02, generalize the encoding methods to support     different prefix lengths.
3.  Align various messages with DAP-02: move the timestamp out of the nonce; define the ReportMetadata structure; update length prefixes; and add "batch-saturated" transition failure type. While at it, add type defs for Time and Duration and fiddle with buffers for AAD computations. (*)
4. Define BatchSelector and Query structs.
5. Add public share to Report and ReportShare. (*)
6. add report count to CollectResp. (*)

(*) Indicates a wire-breaking change.

cjpatton: @chris-wood ready for another round. I updated the size of QueryType to match the spec. I also added a TODO to replace the message encoding stuff with what libprio will ship in 0.10.0.

Patch:
diff --git a/Cargo.lock b/Cargo.lock
index 74cc0dd..1d92666 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -304,9 +304,9 @@ dependencies = [
 
 [[package]]
 name = "clap"
-version = "3.2.20"
+version = "3.2.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "23b71c3ce99b7611011217b366d923f1d0a7e07a92bb2dbf1e84508c673ca3bd"
+checksum = "68d43934757334b5c0519ff882e1ab9647ac0258b47c24c4f490d78e42697fd5"
 dependencies = [
  "atty",
  "bitflags",
@@ -495,7 +495,7 @@ dependencies = [
  "anyhow",
  "assert_matches",
  "base64",
- "clap 3.2.20",
+ "clap 3.2.19",
  "daphne 0.1.2",
  "prio",
  "reqwest",
@@ -508,6 +508,8 @@ dependencies = [
 [[package]]
 name = "daphne"
 version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "150b45b6e1df7fcbaff121c2ac09059152e6335a81e986d89ce5c88abbcba17d"
 dependencies = [
  "assert_matches",
  "async-trait",
@@ -522,15 +524,12 @@ dependencies = [
  "serde",
  "serde_json",
  "thiserror",
- "tokio",
  "url",
 ]
 
 [[package]]
 name = "daphne"
-version = "0.1.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "150b45b6e1df7fcbaff121c2ac09059152e6335a81e986d89ce5c88abbcba17d"
+version = "0.2.0"
 dependencies = [
  "assert_matches",
  "async-trait",
@@ -545,6 +544,7 @@ dependencies = [
  "serde",
  "serde_json",
  "thiserror",
+ "tokio",
  "url",
 ]
 
@@ -556,7 +556,7 @@ dependencies = [
  "base64",
  "cfg-if 0.1.10",
  "console_error_panic_hook",
- "daphne 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "daphne 0.1.2",
  "daphne_worker",
  "deadpool-postgres",
  "futures",
@@ -584,7 +584,7 @@ version = "0.1.2"
 dependencies = [
  "async-trait",
  "base64",
- "daphne 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "daphne 0.1.2",
  "futures",
  "getrandom",
  "hex",
@@ -1632,9 +1632,9 @@ dependencies = [
 
 [[package]]
 name = "once_cell"
-version = "1.14.0"
+version = "1.13.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2f7254b99e31cad77da24b08ebf628882739a608578bb1bcdfc1f9c21260d7c0"
+checksum = "074864da206b4973b84eb91683020dbefd6a8c3f0f38e054d93954e891935e4e"
 
 [[package]]
 name = "opaque-debug"
@@ -2705,9 +2705,9 @@ checksum = "cda74da7e1a664f795bb1f8a87ec406fb89a02522cf6e50620d016add6dbbf5c"
 
 [[package]]
 name = "tokio"
-version = "1.21.0"
+version = "1.20.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "89797afd69d206ccd11fb0ea560a44bbb87731d020670e79416d442919257d42"
+checksum = "7a8325f63a7d4774dd041e363b2409ed1c5cbbd0f867795e661df066b2b0a581"
 dependencies = [
  "autocfg",
  "bytes",
diff --git a/daphne/Cargo.toml b/daphne/Cargo.toml
index e6a65fa..65f03a9 100644
--- a/daphne/Cargo.toml
+++ b/daphne/Cargo.toml
@@ -3,7 +3,7 @@
 [package]
 name = "daphne"
 description = "Implementation of the DAP specification"
-version = "0.1.2"
+version = "0.2.0"
 authors = [
   "Christopher Patton <cpatton@cloudflare.com>",
   "Armando Faz Hernandez <armfazh@cloudflare.com>",
diff --git a/daphne/dapf/Cargo.toml b/daphne/dapf/Cargo.toml
index b9fa392..46ed62c 100644
--- a/daphne/dapf/Cargo.toml
+++ b/daphne/dapf/Cargo.toml
@@ -11,10 +11,10 @@ edition = "2021"
 license = "BSD-3-Clause"
 
 [dependencies]
-daphne = { path = "../" }
+daphne = "0.1.2" # TODO(issue #100) Use path = "../daphne" instead.
 assert_matches = "1.5.0"
 base64 = "0.13.0"
-prio = { version = "=0.8.2", features = ["prio2"] }
+prio = { version = "=0.8.2" }
 serde = { version = "^1.0", features = ["derive"] }
 serde_json = "1.0.82"
 url = { version = "2.2.2", features = ["serde"] }
diff --git a/daphne/src/hpke.rs b/daphne/src/hpke.rs
index 6d3e7df..025b055 100644
--- a/daphne/src/hpke.rs
+++ b/daphne/src/hpke.rs
@@ -5,8 +5,8 @@
 
 use crate::{
     messages::{
-        decode_u16_bytes, encode_u16_bytes, HpkeAeadId, HpkeCiphertext, HpkeConfig, HpkeKdfId,
-        HpkeKemId, Id,
+        decode_prefixed_bytes, encode_prefixed_bytes, HpkeAeadId, HpkeCiphertext, HpkeConfig,
+        HpkeKdfId, HpkeKemId, Id,
     },
     DapError,
 };
@@ -286,7 +286,7 @@ impl<'a> HpkeDecrypter<'a> for HpkeReceiverConfig {
 impl Encode for HpkeReceiverConfig {
     fn encode(&self, bytes: &mut Vec<u8>) {
         self.config.encode(bytes);
-        encode_u16_bytes(bytes, &self.secret_key);
+        encode_prefixed_bytes::<2>(bytes, &self.secret_key);
     }
 }
 
@@ -294,7 +294,7 @@ impl Decode for HpkeReceiverConfig {
     fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
         Ok(Self {
             config: HpkeConfig::decode(bytes)?,
-            secret_key: decode_u16_bytes(bytes)?,
+            secret_key: decode_prefixed_bytes::<2>(bytes)?,
         })
     }
 }
diff --git a/daphne/src/lib.rs b/daphne/src/lib.rs
index 9a07a2b..583fad5 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, Interval, Nonce, TransitionFailure},
+    messages::{CollectResp, HpkeConfig, Nonce, Time, TransitionFailure},
     vdaf::{
         prio2::prio2_decode_prepare_state,
         prio3::{prio3_append_prepare_state, prio3_decode_prepare_state},
@@ -313,14 +313,17 @@ pub struct DapTaskConfig {
     pub(crate) collector_hpke_config: HpkeConfig,
 }
 
+#[cfg(test)]
 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 current_batch_window(&self, now: u64) -> Interval {
+    pub fn query_for_current_batch_window(&self, now: u64) -> crate::messages::Query {
         let start = now - (now % self.min_batch_duration);
-        Interval {
-            start,
-            duration: self.min_batch_duration,
+        crate::messages::Query::TimeInterval {
+            batch_interval: crate::messages::Interval {
+                start,
+                duration: self.min_batch_duration,
+            },
         }
     }
 }
@@ -386,7 +389,7 @@ pub enum DapAggregateResult {
 /// The Leader's state after sending an AggregateInitReq.
 #[derive(Debug)]
 pub struct DapLeaderState {
-    pub(crate) seq: Vec<(VdafState, VdafMessage, Nonce)>,
+    pub(crate) seq: Vec<(VdafState, VdafMessage, Time, Nonce)>,
 }
 
 /// The Leader's state after sending an AggregateContReq.
@@ -398,7 +401,7 @@ pub struct DapLeaderUncommitted {
 /// The Helper's state during the aggregation flow.
 #[derive(Clone, Debug, PartialEq)]
 pub struct DapHelperState {
-    pub(crate) seq: Vec<(VdafState, Nonce)>,
+    pub(crate) seq: Vec<(VdafState, Time, Nonce)>,
 }
 
 impl DapHelperState {
@@ -411,7 +414,7 @@ impl DapHelperState {
     /// Note that the encoding format is not specified by the DAP standard.
     pub fn get_encoded(&self, vdaf_config: &VdafConfig) -> Result<Vec<u8>, DapError> {
         let mut bytes = vec![];
-        for (state, nonce) in self.seq.iter() {
+        for (state, time, nonce) in self.seq.iter() {
             match (vdaf_config, state) {
                 (VdafConfig::Prio3(prio3_config), _) => {
                     prio3_append_prepare_state(&mut bytes, prio3_config, state)?;
@@ -421,6 +424,7 @@ impl DapHelperState {
                 }
                 _ => return Err(DapError::fatal("VDAF config and prep state mismatch")),
             }
+            time.encode(&mut bytes);
             nonce.encode(&mut bytes);
         }
         Ok(bytes)
@@ -439,8 +443,9 @@ impl DapHelperState {
                     prio2_decode_prepare_state(*dimension, 1, &mut r)?
                 }
             };
+            let time = Time::decode(&mut r)?;
             let nonce = Nonce::decode(&mut r)?;
-            seq.push((state, nonce))
+            seq.push((state, time, nonce))
         }
 
         Ok(DapHelperState { seq })
diff --git a/daphne/src/messages.rs b/daphne/src/messages.rs
index 5f90529..5b1d08d 100644
--- a/daphne/src/messages.rs
+++ b/daphne/src/messages.rs
@@ -4,7 +4,7 @@
 //! Messages in the DAP protocol.
 
 use crate::DapTaskConfig;
-use prio::codec::{decode_u16_items, encode_u16_items, CodecError, Decode, Encode};
+use prio::codec::{CodecError, Decode, Encode};
 use serde::{Deserialize, Serialize};
 use std::{
     convert::{TryFrom, TryInto},
@@ -16,6 +16,8 @@ const KEM_ID_X25519_HKDF_SHA256: u16 = 0x0020;
 const KEM_ID_P256_HKDF_SHA256: u16 = 0x0010;
 const KDF_ID_HKDF_SHA256: u16 = 0x0001;
 const AEAD_ID_AES128GCM: u16 = 0x0001;
+const QUERY_TYPE_TIME_INTERVAL: u16 = 0x0001;
+const QUERY_TYPE_FIXED_SIZE: u16 = 0x0002;
 
 /// The identifier for a DAP task.
 #[derive(Clone, Debug, Default, Deserialize, Hash, PartialEq, Eq, Serialize)]
@@ -53,27 +55,34 @@ impl AsRef<[u8]> for Id {
     }
 }
 
-/// The timestamp and random number fields of a [`Report`].
+/// A duration.
+pub type Duration = u64;
+
+/// The timestamp sent in a [`Report`].
+pub type Time = u64;
+
+/// The nonce sent in a [`Report`].
 #[derive(Clone, Debug, Deserialize, PartialEq, Eq, Hash, Serialize)]
 #[allow(missing_docs)]
-pub struct Nonce {
-    pub time: u64,
-    pub rand: [u8; 16],
-}
+pub struct Nonce(pub [u8; 16]);
 
 impl Encode for Nonce {
     fn encode(&self, bytes: &mut Vec<u8>) {
-        self.time.encode(bytes);
-        bytes.extend_from_slice(&self.rand);
+        bytes.extend_from_slice(&self.0);
     }
 }
 
 impl Decode for Nonce {
     fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
-        let time = u64::decode(bytes)?;
-        let mut rand = [0; 16];
-        bytes.read_exact(&mut rand)?;
-        Ok(Nonce { time, rand })
+        let mut nonce = [0; 16];
+        bytes.read_exact(&mut nonce)?;
+        Ok(Nonce(nonce))
+    }
+}
+
+impl AsRef<[u8]> for Nonce {
+    fn as_ref(&self) -> &[u8] {
+        &self.0
     }
 }
 
@@ -88,7 +97,7 @@ impl Encode for Extension {
         match self {
             Self::Unhandled { typ, payload } => {
                 typ.encode(bytes);
-                encode_u16_bytes(bytes, payload);
+                encode_prefixed_bytes::<2>(bytes, payload);
             }
         }
     }
@@ -98,7 +107,34 @@ impl Decode for Extension {
     fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
         Ok(Self::Unhandled {
             typ: u16::decode(bytes)?,
-            payload: decode_u16_bytes(bytes)?,
+            payload: decode_prefixed_bytes::<2>(bytes)?,
+        })
+    }
+}
+
+/// Report metadata.
+#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
+#[allow(missing_docs)]
+pub struct ReportMetadata {
+    pub time: Time,
+    pub nonce: Nonce,
+    pub extensions: Vec<Extension>,
+}
+
+impl Encode for ReportMetadata {
+    fn encode(&self, bytes: &mut Vec<u8>) {
+        self.time.encode(bytes);
+        self.nonce.encode(bytes);
+        encode_prefixed_items::<2, _>(bytes, &self.extensions);
+    }
+}
+
+impl Decode for ReportMetadata {
+    fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
+        Ok(Self {
+            time: Time::decode(bytes)?,
+            nonce: Nonce::decode(bytes)?,
+            extensions: decode_prefixed_items::<2, _>(bytes)?,
         })
     }
 }
@@ -108,17 +144,17 @@ impl Decode for Extension {
 #[allow(missing_docs)]
 pub struct Report {
     pub task_id: Id,
-    pub nonce: Nonce,
-    pub extensions: Vec<Extension>,
+    pub metadata: ReportMetadata,
+    pub public_share: Vec<u8>,
     pub encrypted_input_shares: Vec<HpkeCiphertext>,
 }
 
 impl Encode for Report {
     fn encode(&self, bytes: &mut Vec<u8>) {
         self.task_id.encode(bytes);
-        self.nonce.encode(bytes);
-        encode_u16_items(bytes, &(), &self.extensions);
-        encode_u16_items(bytes, &(), &self.encrypted_input_shares);
+        self.metadata.encode(bytes);
+        encode_prefixed_bytes::<4>(bytes, &self.public_share);
+        encode_prefixed_items::<4, _>(bytes, &self.encrypted_input_shares);
     }
 }
 
@@ -126,9 +162,9 @@ impl Decode for Report {
     fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
         Ok(Self {
             task_id: Id::decode(bytes)?,
-            nonce: Nonce::decode(bytes)?,
-            extensions: decode_u16_items(&(), bytes)?,
-            encrypted_input_shares: decode_u16_items(&(), bytes)?,
+            metadata: ReportMetadata::decode(bytes)?,
+            public_share: decode_prefixed_bytes::<4>(bytes)?,
+            encrypted_input_shares: decode_prefixed_items::<4, _>(bytes)?,
         })
     }
 }
@@ -138,15 +174,15 @@ impl Decode for Report {
 #[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
 #[allow(missing_docs)]
 pub struct ReportShare {
-    pub nonce: Nonce,
-    pub extensions: Vec<Extension>,
+    pub metadata: ReportMetadata,
+    pub public_share: Vec<u8>,
     pub encrypted_input_share: HpkeCiphertext,
 }
 
 impl Encode for ReportShare {
     fn encode(&self, bytes: &mut Vec<u8>) {
-        self.nonce.encode(bytes);
-        encode_u16_items(bytes, &(), &self.extensions);
+        self.metadata.encode(bytes);
+        encode_prefixed_bytes::<4>(bytes, &self.public_share);
         self.encrypted_input_share.encode(bytes);
     }
 }
@@ -154,19 +190,50 @@ impl Encode for ReportShare {
 impl Decode for ReportShare {
     fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
         Ok(Self {
-            nonce: Nonce::decode(bytes)?,
-            extensions: decode_u16_items(&(), bytes)?,
+            metadata: ReportMetadata::decode(bytes)?,
+            public_share: decode_prefixed_bytes::<4>(bytes)?,
             encrypted_input_share: HpkeCiphertext::decode(bytes)?,
         })
     }
 }
 
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum BatchParameter {
+    TimeInterval,
+    FixedSize { batch_id: Id },
+}
+
+impl Encode for BatchParameter {
+    fn encode(&self, bytes: &mut Vec<u8>) {
+        match self {
+            Self::TimeInterval => QUERY_TYPE_TIME_INTERVAL.encode(bytes),
+            Self::FixedSize { batch_id } => {
+                QUERY_TYPE_FIXED_SIZE.encode(bytes);
+                batch_id.encode(bytes);
+            }
+        }
+    }
+}
+
+impl Decode for BatchParameter {
+    fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
+        match u16::decode(bytes)? {
+            QUERY_TYPE_TIME_INTERVAL => Ok(Self::TimeInterval),
+            QUERY_TYPE_FIXED_SIZE => Ok(Self::FixedSize {
+                batch_id: Id::decode(bytes)?,
+            }),
+            _ => Err(CodecError::UnexpectedValue),
+        }
+    }
+}
+
 /// Aggregate initialization request.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct AggregateInitializeReq {
     pub task_id: Id,
     pub agg_job_id: Id,
     pub agg_param: Vec<u8>,
+    pub batch_param: BatchParameter,
     pub report_shares: Vec<ReportShare>,
 }
 
@@ -174,8 +241,9 @@ impl Encode for AggregateInitializeReq {
     fn encode(&self, bytes: &mut Vec<u8>) {
         self.task_id.encode(bytes);
         self.agg_job_id.encode(bytes);
-        encode_u16_bytes(bytes, &self.agg_param);
-        encode_u16_items(bytes, &(), &self.report_shares);
+        encode_prefixed_bytes::<2>(bytes, &self.agg_param);
+        self.batch_param.encode(bytes);
+        encode_prefixed_items::<4, _>(bytes, &self.report_shares);
     }
 }
 
@@ -184,8 +252,9 @@ impl Decode for AggregateInitializeReq {
         Ok(Self {
             task_id: Id::decode(bytes)?,
             agg_job_id: Id::decode(bytes)?,
-            agg_param: decode_u16_bytes(bytes)?,
-            report_shares: decode_u16_items(&(), bytes)?,
+            agg_param: decode_prefixed_bytes::<2>(bytes)?,
+            batch_param: BatchParameter::decode(bytes)?,
+            report_shares: decode_prefixed_items::<4, _>(bytes)?,
         })
     }
 }
@@ -202,7 +271,7 @@ impl Encode for AggregateContinueReq {
     fn encode(&self, bytes: &mut Vec<u8>) {
         self.task_id.encode(bytes);
         self.agg_job_id.encode(bytes);
-        encode_u16_items(bytes, &(), &self.transitions);
+        encode_prefixed_items::<4, _>(bytes, &self.transitions);
     }
 }
 
@@ -211,7 +280,7 @@ impl Decode for AggregateContinueReq {
         Ok(Self {
             task_id: Id::decode(bytes)?,
             agg_job_id: Id::decode(bytes)?,
-            transitions: decode_u16_items(&(), bytes)?,
+            transitions: decode_prefixed_items::<4, _>(bytes)?,
         })
     }
 }
@@ -256,7 +325,7 @@ impl Encode for TransitionVar {
         match self {
             TransitionVar::Continued(vdaf_message) => {
                 0_u8.encode(bytes);
-                encode_u16_bytes(bytes, vdaf_message);
+                encode_prefixed_bytes::<4>(bytes, vdaf_message);
             }
             TransitionVar::Finished => {
                 1_u8.encode(bytes);
@@ -272,7 +341,7 @@ impl Encode for TransitionVar {
 impl Decode for TransitionVar {
     fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
         match u8::decode(bytes)? {
-            0 => Ok(Self::Continued(decode_u16_bytes(bytes)?)),
+            0 => Ok(Self::Continued(decode_prefixed_bytes::<4>(bytes)?)),
             1 => Ok(Self::Finished),
             2 => Ok(Self::Failed(TransitionFailure::decode(bytes)?)),
             _ => Err(CodecError::UnexpectedValue),
@@ -289,6 +358,7 @@ pub enum TransitionFailure {
     HpkeUnknownConfigId = 3,
     HpkeDecryptError = 4,
     VdafPrepError = 5,
+    BatchSaturated = 6,
 }
 
 impl TryFrom<u8> for TransitionFailure {
@@ -302,6 +372,7 @@ impl TryFrom<u8> for TransitionFailure {
             b if b == Self::HpkeUnknownConfigId as u8 => Ok(Self::HpkeUnknownConfigId),
             b if b == Self::HpkeDecryptError as u8 => Ok(Self::HpkeDecryptError),
             b if b == Self::VdafPrepError as u8 => Ok(Self::VdafPrepError),
+            b if b == Self::BatchSaturated as u8 => Ok(Self::BatchSaturated),
             _ => Err(CodecError::UnexpectedValue),
         }
     }
@@ -328,6 +399,7 @@ impl std::fmt::Display for TransitionFailure {
             Self::HpkeUnknownConfigId => write!(f, "hpke-unknown-config-id({})", *self as u8),
             Self::HpkeDecryptError => write!(f, "hpke-decrypt-error({})", *self as u8),
             Self::VdafPrepError => write!(f, "vdaf-prep-error({})", *self as u8),
+            Self::BatchSaturated => write!(f, "batch-saturated({})", *self as u8),
         }
     }
 }
@@ -341,14 +413,14 @@ pub struct AggregateResp {
 
 impl Encode for AggregateResp {
     fn encode(&self, bytes: &mut Vec<u8>) {
-        encode_u16_items(bytes, &(), &self.transitions);
+        encode_prefixed_items::<4, _>(bytes, &self.transitions);
     }
 }
 
 impl Decode for AggregateResp {
     fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
         Ok(Self {
-            transitions: decode_u16_items(&(), bytes)?,
+            transitions: decode_prefixed_items::<4, _>(bytes)?,
         })
     }
 }
@@ -357,13 +429,13 @@ impl Decode for AggregateResp {
 #[derive(Clone, Debug, Default, Deserialize, PartialEq, Eq, Serialize)]
 #[allow(missing_docs)]
 pub struct Interval {
-    pub start: u64,
-    pub duration: u64,
+    pub start: Time,
+    pub duration: Duration,
 }
 
 impl Interval {
     /// Return the end of the interval, i.e., `self.start + self.duration`.
-    pub fn end(&self) -> u64 {
+    pub fn end(&self) -> Time {
         self.start + self.duration
     }
 
@@ -389,27 +461,82 @@ impl Encode for Interval {
 impl Decode for Interval {
     fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
         Ok(Self {
-            start: u64::decode(bytes)?,
-            duration: u64::decode(bytes)?,
+            start: Time::decode(bytes)?,
+            duration: Duration::decode(bytes)?,
         })
     }
 }
 
+/// A query issued by the Collector in a collect request.
+#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum Query {
+    TimeInterval { batch_interval: Interval },
+    FixedSize { batch_id: Id },
+}
+
+impl Query {
+    pub(crate) fn unwrap_interval(&self) -> &Interval {
+        if let BatchSelector::TimeInterval { ref batch_interval } = self {
+            batch_interval
+        } else {
+            panic!("TODO(issue #100)");
+        }
+    }
+}
+
+impl Encode for Query {
+    fn encode(&self, bytes: &mut Vec<u8>) {
+        match self {
+            Self::TimeInterval { batch_interval } => {
+                QUERY_TYPE_TIME_INTERVAL.encode(bytes);
+                batch_interval.encode(bytes);
+            }
+            Self::FixedSize { batch_id } => {
+                QUERY_TYPE_FIXED_SIZE.encode(bytes);
+                batch_id.encode(bytes);
+            }
+        }
+    }
+}
+
+impl Decode for Query {
+    fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
+        match u16::decode(bytes)? {
+            QUERY_TYPE_TIME_INTERVAL => Ok(Self::TimeInterval {
+                batch_interval: Interval::decode(bytes)?,
+            }),
+            QUERY_TYPE_FIXED_SIZE => Ok(Self::FixedSize {
+                batch_id: Id::decode(bytes)?,
+            }),
+            _ => Err(CodecError::UnexpectedValue),
+        }
+    }
+}
+
+impl Default for Query {
+    fn default() -> Self {
+        Self::TimeInterval {
+            batch_interval: Interval::default(),
+        }
+    }
+}
+
 /// A collect request.
 //
 // TODO Add serialization tests.
 #[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
 pub struct CollectReq {
     pub task_id: Id,
-    pub batch_interval: Interval,
+    pub query: Query,
     pub agg_param: Vec<u8>,
 }
 
 impl Encode for CollectReq {
     fn encode(&self, bytes: &mut Vec<u8>) {
         self.task_id.encode(bytes);
-        self.batch_interval.encode(bytes);
-        encode_u16_bytes(bytes, &self.agg_param);
+        self.query.encode(bytes);
+        encode_prefixed_bytes::<2>(bytes, &self.agg_param);
     }
 }
 
@@ -417,8 +544,8 @@ impl Decode for CollectReq {
     fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
         Ok(Self {
             task_id: Id::decode(bytes)?,
-            batch_interval: Interval::decode(bytes)?,
-            agg_param: decode_u16_bytes(bytes)?,
+            query: Query::decode(bytes)?,
+            agg_param: decode_prefixed_bytes::<2>(bytes)?,
         })
     }
 }
@@ -428,30 +555,40 @@ impl Decode for CollectReq {
 // TODO Add serialization tests.
 #[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
 pub struct CollectResp {
+    pub report_count: u64,
     pub encrypted_agg_shares: Vec<HpkeCiphertext>,
 }
 
 impl Encode for CollectResp {
     fn encode(&self, bytes: &mut Vec<u8>) {
-        encode_u16_items(bytes, &(), &self.encrypted_agg_shares);
+        self.report_count.encode(bytes);
+        encode_prefixed_items::<4, _>(bytes, &self.encrypted_agg_shares);
     }
 }
 
 impl Decode for CollectResp {
     fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
         Ok(Self {
-            encrypted_agg_shares: decode_u16_items(&(), bytes)?,
+            report_count: u64::decode(bytes)?,
+            encrypted_agg_shares: decode_prefixed_items::<4, _>(bytes)?,
         })
     }
 }
 
+/// A batch selector issued by the Leader in an aggregate-share request.
+//
+// NOTE(cjpatton) This structure is identical to Query, hence the typedef. Eventually the
+// strudcture might change (https://github.com/ietf-wg-ppm/draft-ietf-ppm-dap/issues/342), at which
+// point we will need to make BatchSelector its own struct.
+pub type BatchSelector = Query;
+
 /// An aggregate-share request.
 //
 // TODO Add serialization tests.
 #[derive(Debug, Default)]
 pub struct AggregateShareReq {
     pub task_id: Id,
-    pub batch_interval: Interval,
+    pub batch_selector: BatchSelector,
     pub agg_param: Vec<u8>,
     pub report_count: u64,
     pub checksum: [u8; 32],
@@ -460,8 +597,8 @@ pub struct AggregateShareReq {
 impl Encode for AggregateShareReq {
     fn encode(&self, bytes: &mut Vec<u8>) {
         self.task_id.encode(bytes);
-        self.batch_interval.encode(bytes);
-        encode_u16_bytes(bytes, &self.agg_param);
+        self.batch_selector.encode(bytes);
+        encode_prefixed_bytes::<2>(bytes, &self.agg_param);
         self.report_count.encode(bytes);
         bytes.extend_from_slice(&self.checksum);
     }
@@ -471,8 +608,8 @@ impl Decode for AggregateShareReq {
     fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
         Ok(Self {
             task_id: Id::decode(bytes)?,
-            batch_interval: Interval::decode(bytes)?,
-            agg_param: decode_u16_bytes(bytes)?,
+            batch_selector: BatchSelector::decode(bytes)?,
+            agg_param: decode_prefixed_bytes::<2>(bytes)?,
             report_count: u64::decode(bytes)?,
             checksum: {
                 let mut checksum = [0u8; 32];
@@ -628,7 +765,7 @@ impl Encode for HpkeConfig {
         self.kem_id.encode(bytes);
         self.kdf_id.encode(bytes);
         self.aead_id.encode(bytes);
-        encode_u16_bytes(bytes, &self.public_key);
+        encode_prefixed_bytes::<2>(bytes, &self.public_key);
     }
 }
 
@@ -639,7 +776,7 @@ impl Decode for HpkeConfig {
             kem_id: HpkeKemId::decode(bytes)?,
             kdf_id: HpkeKdfId::decode(bytes)?,
             aead_id: HpkeAeadId::decode(bytes)?,
-            public_key: decode_u16_bytes(bytes)?,
+            public_key: decode_prefixed_bytes::<2>(bytes)?,
         })
     }
 }
@@ -659,8 +796,8 @@ pub struct HpkeCiphertext {
 impl Encode for HpkeCiphertext {
     fn encode(&self, bytes: &mut Vec<u8>) {
         self.config_id.encode(bytes);
-        encode_u16_bytes(bytes, &self.enc);
-        encode_u16_bytes(bytes, &self.payload);
+        encode_prefixed_bytes::<2>(bytes, &self.enc);
+        encode_prefixed_bytes::<4>(bytes, &self.payload);
     }
 }
 
@@ -668,8 +805,8 @@ impl Decode for HpkeCiphertext {
     fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
         Ok(Self {
             config_id: u8::decode(bytes)?,
-            enc: decode_u16_bytes(bytes)?,
-            payload: decode_u16_bytes(bytes)?,
+            enc: decode_prefixed_bytes::<2>(bytes)?,
+            payload: decode_prefixed_bytes::<4>(bytes)?,
         })
     }
 }
@@ -688,15 +825,83 @@ pub(crate) fn constant_time_eq(left: &[u8], right: &[u8]) -> bool {
     r == 0
 }
 
-pub(crate) fn encode_u16_bytes(bytes: &mut Vec<u8>, input: &[u8]) {
-    let len: u16 = input.len().try_into().unwrap();
-    len.encode(bytes);
+// TODO(issue #100) Replace this stuff with libprio once 0.10.0 is ready.
+fn encode_prefix<const PREFIX_BYTES: usize>(bytes: &mut Vec<u8>, len: usize) {
+    let prefix_start = bytes.len();
+    bytes.extend_from_slice(&[0; PREFIX_BYTES]);
+    let prefix_end = bytes.len();
+    write_prefix::<PREFIX_BYTES>(&mut bytes[prefix_start..prefix_end], len);
+}
+
+fn write_prefix<const PREFIX_BYTES: usize>(buf: &mut [u8], len: usize) {
+    for i in (0..PREFIX_BYTES).rev() {
+        buf[PREFIX_BYTES - i - 1] = ((len >> (i << 3)) & 255) as u8;
+    }
+}
+
+pub(crate) fn decode_prefix<const PREFIX_BYTES: usize>(
+    bytes: &mut Cursor<&[u8]>,
+) -> Result<usize, CodecError> {
+    let mut len = 0;
+    for i in (0..PREFIX_BYTES).rev() {
+        len |= (u8::decode(bytes)? as usize) << (i << 3);
+    }
+    Ok(len)
+}
+
+pub(crate) fn encode_prefixed_bytes<const PREFIX_BYTES: usize>(bytes: &mut Vec<u8>, input: &[u8]) {
+    encode_prefix::<PREFIX_BYTES>(bytes, input.len());
     bytes.extend_from_slice(input);
 }
 
-pub(crate) fn decode_u16_bytes(bytes: &mut Cursor<&[u8]>) -> Result<Vec<u8>, CodecError> {
-    let len = u16::decode(bytes)? as usize;
+pub(crate) fn decode_prefixed_bytes<const PREFIX_BYTES: usize>(
+    bytes: &mut Cursor<&[u8]>,
+) -> Result<Vec<u8>, CodecError> {
+    let len = decode_prefix::<PREFIX_BYTES>(bytes)?;
     let mut out = vec![0; len];
     bytes.read_exact(&mut out)?;
     Ok(out)
 }
+
+fn encode_prefixed_items<const PREFIX_BYTES: usize, I: Encode>(bytes: &mut Vec<u8>, items: &[I]) {
+    let prefix_start = bytes.len();
+    bytes.extend_from_slice(&[0; PREFIX_BYTES]);
+    let prefix_end = bytes.len();
+    for item in items {
+        item.encode(bytes);
+    }
+    let len = bytes.len() - prefix_end;
+    write_prefix::<PREFIX_BYTES>(&mut bytes[prefix_start..prefix_end], len);
+}
+
+fn decode_prefixed_items<const PREFIX_BYTES: usize, I: Decode>(
+    bytes: &mut Cursor<&[u8]>,
+) -> Result<Vec<I>, CodecError> {
+    let len = decode_prefix::<PREFIX_BYTES>(bytes)?;
+    decode_items(len, bytes)
+}
+
+fn decode_items<I: Decode>(len: usize, bytes: &mut Cursor<&[u8]>) -> Result<Vec<I>, CodecError> {
+    let mut decoded = Vec::new();
+    let initial_position = bytes.position() as usize;
+
+    // Create cursor over specified portion of provided cursor to ensure we can't read past length.
+    let inner = bytes.get_ref();
+
+    // Make sure encoded length doesn't overflow usize or go past the end of provided byte buffer.
+    let (items_end, overflowed) = initial_position.overflowing_add(len);
+    if overflowed || items_end > inner.len() {
+        return Err(CodecError::LengthPrefixTooBig(len));
+    }
+
+    let mut sub = Cursor::new(&bytes.get_ref()[initial_position..items_end]);
+
+    while sub.position() < len as u64 {
+        decoded.push(I::decode(&mut sub)?);
+    }
+
+    // Advance outer cursor by the amount read in the inner cursor
+    bytes.set_position(initial_position as u64 + sub.position());
+
+    Ok(decoded)
+}
diff --git a/daphne/src/roles.rs b/daphne/src/roles.rs
index 54b761d..0232b42 100644
--- a/daphne/src/roles.rs
+++ b/daphne/src/roles.rs
@@ -12,7 +12,7 @@ use crate::{
     hpke::HpkeDecrypter,
     messages::{
         constant_time_eq, AggregateContinueReq, AggregateInitializeReq, AggregateResp,
-        AggregateShareReq, AggregateShareResp, CollectReq, CollectResp, Id, Interval, Nonce,
+        AggregateShareReq, AggregateShareResp, BatchSelector, CollectReq, CollectResp, Id, Nonce,
         Report, ReportShare, TransitionFailure, TransitionVar,
     },
     DapAbort, DapAggregateShare, DapCollectJob, DapError, DapGlobalConfig, DapHelperState,
@@ -28,10 +28,11 @@ use url::Url;
 macro_rules! check_batch_param {
     (
         $task_config:expr,
-        $batch_interval:expr,
+        $batch_selector:expr,
         $agg_param:expr
     ) => {{
-        if !$batch_interval.is_valid_for($task_config) {
+        let batch_interval = $batch_selector.unwrap_interval();
+        if !batch_interval.is_valid_for($task_config) {
             return Err(DapAbort::InvalidBatchInterval);
         }
 
@@ -43,21 +44,22 @@ macro_rules! check_batch_param {
     }};
 }
 
-macro_rules! check_batch_interval_boundary {
+macro_rules! check_batch_boundary {
     (
         $global_config:expr,
         $now:expr,
-        $batch_interval:expr
+        $batch_selector:expr
     ) => {{
-        if $batch_interval.duration > $global_config.max_batch_duration {
+        let batch_interval = $batch_selector.unwrap_interval();
+        if batch_interval.duration > $global_config.max_batch_duration {
             return Err(DapAbort::BadRequest("batch interval too large".to_string()));
         }
-        if $now.abs_diff($batch_interval.start) > $global_config.min_batch_interval_start {
+        if $now.abs_diff(batch_interval.start) > $global_config.min_batch_interval_start {
             return Err(DapAbort::BadRequest(
                 "batch interval too far into past".to_string(),
             ));
         }
-        if $now.abs_diff($batch_interval.end()) > $global_config.max_batch_interval_end {
+        if $now.abs_diff(batch_interval.end()) > $global_config.max_batch_interval_end {
             return Err(DapAbort::BadRequest(
                 "batch interval too far into future".to_string(),
             ));
@@ -98,12 +100,12 @@ pub trait DapAggregator<'a, S>: HpkeDecrypter<'a> + Sized {
     /// Get the current time (number of seconds since the beginning of UNIX time).
     fn get_current_time(&self) -> u64;
 
-    /// Check whether the given collect request interval does not overlap
-    /// with requests received in the past.
+    /// Check whether the batch determined by the collect request would overlap with a previous
+    /// batch.
     async fn is_batch_overlapping(
         &self,
         task_id: &Id,
-        batch_interval: &Interval,
+        batch_selector: &BatchSelector,
     ) -> Result<bool, DapError>;
 
     /// Store a set of output shares.
@@ -113,16 +115,19 @@ pub trait DapAggregator<'a, S>: HpkeDecrypter<'a> + Sized {
         out_shares: Vec<DapOutputShare>,
     ) -> Result<(), DapError>;
 
-    /// Fetch the aggregate share for the given batch interval.
+    /// Fetch the aggregate share for the given batch.
     async fn get_agg_share(
         &self,
         task_id: &Id,
-        batch_interval: &Interval,
+        batch_selector: &BatchSelector,
     ) -> Result<DapAggregateShare, DapError>;
 
     /// Mark a batch as collected.
-    async fn mark_collected(&self, task_id: &Id, batch_interval: &Interval)
-        -> Result<(), DapError>;
+    async fn mark_collected(
+        &self,
+        task_id: &Id,
+        batch_selector: &BatchSelector,
+    ) -> Result<(), DapError>;
 
     /// Handle HTTP GET to `/hpke_config?task_id=<task_id>`.
     async fn http_get_hpke_config(&'a self, req: &DapRequest<S>) -> Result<DapResponse, DapAbort> {
@@ -297,31 +302,28 @@ pub trait DapLeader<'a, S>: DapAuthorizedSender<S> + DapAggregator<'a, S> {
             .await?
             .ok_or(DapAbort::UnrecognizedTask)?;
         let task_config = wrapped_task_config.as_ref();
-        check_batch_param!(
-            task_config,
-            collect_req.batch_interval,
-            collect_req.agg_param
-        );
+        check_batch_param!(task_config, collect_req.query, collect_req.agg_param);
 
         // Check whether the DAP version in the request matches the task config.
         if task_config.version != req.version {
             return Err(DapAbort::InvalidProtocolVersion);
         }
 
-        // Check that the batch interval is not too wide or too far into the past/future.
+        // Check that the batch boundaries.
         let global_config = self.get_global_config();
         let now = self.get_current_time();
-        check_batch_interval_boundary!(global_config, now, collect_req.batch_interval);
+        check_batch_boundary!(global_config, now, collect_req.query);
 
-        // Check that the batch interval does not overlap with past requests.
+        // Check that the batch does not overlap with previous batches.
         if self
-            .is_batch_overlapping(&collect_req.task_id, &collect_req.batch_interval)
+            .is_batch_overlapping(&collect_req.task_id, &collect_req.query)
             .await?
         {
             return Err(DapAbort::BatchOverlap);
         }
 
-        if !collect_req.batch_interval.is_valid_for(task_config) {
+        let batch_interval = collect_req.query.unwrap_interval();
+        if !batch_interval.is_valid_for(task_config) {
             return Err(DapAbort::InvalidBatchInterval);
         }
 
@@ -424,7 +426,7 @@ pub trait DapLeader<'a, S>: DapAuthorizedSender<S> + DapAggregator<'a, S> {
         collect_req: CollectReq,
     ) -> Result<u64, DapAbort> {
         let leader_agg_share = self
-            .get_agg_share(&collect_req.task_id, &collect_req.batch_interval)
+            .get_agg_share(&collect_req.task_id, &collect_req.query)
             .await?;
 
         // Check that the minimum batch size is met.
@@ -436,14 +438,14 @@ pub trait DapLeader<'a, S>: DapAuthorizedSender<S> + DapAggregator<'a, S> {
         let leader_enc_agg_share = task_config.vdaf.produce_leader_encrypted_agg_share(
             &task_config.collector_hpke_config,
             &collect_req.task_id,
-            &collect_req.batch_interval,
+            &collect_req.query,
             &leader_agg_share,
         )?;
 
         // Prepare AggregateShareReq.
         let agg_share_req = AggregateShareReq {
             task_id: collect_req.task_id.clone(),
-            batch_interval: collect_req.batch_interval.clone(),
+            batch_selector: collect_req.query.clone(),
             agg_param: collect_req.agg_param.clone(),
             report_count: leader_agg_share.report_count,
             checksum: leader_agg_share.checksum,
@@ -462,13 +464,14 @@ pub trait DapLeader<'a, S>: DapAuthorizedSender<S> + DapAggregator<'a, S> {
 
         // Complete the collect job.
         let collect_resp = CollectResp {
+            report_count: leader_agg_share.report_count,
             encrypted_agg_shares: vec![leader_enc_agg_share, agg_share_resp.encrypted_agg_share],
         };
         self.finish_collect_job(&collect_req.task_id, collect_id, &collect_resp)
             .await?;
 
         // Mark reports as collected.
-        self.mark_collected(&agg_share_req.task_id, &agg_share_req.batch_interval)
+        self.mark_collected(&agg_share_req.task_id, &agg_share_req.batch_selector)
             .await?;
 
         Ok(agg_share_req.report_count)
@@ -613,7 +616,7 @@ pub trait DapHelper<'a, S>: DapAggregator<'a, S> {
                                 agg_resp.transitions[i].var = TransitionVar::Failed(*failure);
 
                                 // Remove VDAF preparation state of reports that were rejected early.
-                                if state.seq[i].1 == agg_resp.transitions[i].nonce {
+                                if state.seq[i].2 == agg_resp.transitions[i].nonce {
                                     let _val = state.seq.remove(i);
                                 } else {
                                     // The nonce in the Helper state and Aggregate response must be aligned.
@@ -708,7 +711,7 @@ pub trait DapHelper<'a, S>: DapAggregator<'a, S> {
         let task_config = wrapped_task_config.as_ref();
         check_batch_param!(
             task_config,
-            agg_share_req.batch_interval,
+            agg_share_req.batch_selector,
             agg_share_req.agg_param
         );
 
@@ -717,21 +720,20 @@ pub trait DapHelper<'a, S>: DapAggregator<'a, S> {
             return Err(DapAbort::InvalidProtocolVersion);
         }
 
-        // Check that the batch interval is not too wide or too far into the past/future.
+        // Check the batch boundaries.
         let global_config = self.get_global_config();
         let now = self.get_current_time();
-        check_batch_interval_boundary!(global_config, now, agg_share_req.batch_interval);
+        check_batch_boundary!(global_config, now, agg_share_req.batch_selector);
 
-        // Check that the batch interval does not overlap with past requests.
+        // Check that the batch does not overlap with previous boundaries.
         if self
-            .is_batch_overlapping(&agg_share_req.task_id, &agg_share_req.batch_interval)
+            .is_batch_overlapping(&agg_share_req.task_id, &agg_share_req.batch_selector)
             .await?
         {
             return Err(DapAbort::BatchOverlap);
         }
-
         let agg_share = self
-            .get_agg_share(&agg_share_req.task_id, &agg_share_req.batch_interval)
+            .get_agg_share(&agg_share_req.task_id, &agg_share_req.batch_selector)
             .await?;
 
         // Check that we have aggreagted the same set of reports as the leader.
@@ -747,13 +749,13 @@ pub trait DapHelper<'a, S>: DapAggregator<'a, S> {
         }
 
         // Mark each aggregated report as collected.
-        self.mark_collected(&agg_share_req.task_id, &agg_share_req.batch_interval)
+        self.mark_collected(&agg_share_req.task_id, &agg_share_req.batch_selector)
             .await?;
 
         let encrypted_agg_share = task_config.vdaf.produce_helper_encrypted_agg_share(
             &task_config.collector_hpke_config,
             &agg_share_req.task_id,
-            &agg_share_req.batch_interval,
+            &agg_share_req.batch_selector,
             &agg_share,
         )?;
 
diff --git a/daphne/src/vdaf/mod.rs b/daphne/src/vdaf/mod.rs
index 705f6fa..0aebe44 100644
--- a/daphne/src/vdaf/mod.rs
+++ b/daphne/src/vdaf/mod.rs
@@ -7,9 +7,9 @@
 use crate::{
     hpke::HpkeDecrypter,
     messages::{
-        AggregateContinueReq, AggregateInitializeReq, AggregateResp, Extension, HpkeCiphertext,
-        HpkeConfig, Id, Interval, Nonce, Report, ReportShare, Transition, TransitionFailure,
-        TransitionVar,
+        encode_prefixed_bytes, AggregateContinueReq, AggregateInitializeReq, AggregateResp,
+        BatchParameter, BatchSelector, HpkeCiphertext, HpkeConfig, Id, Nonce, Report,
+        ReportMetadata, ReportShare, Time, Transition, TransitionFailure, TransitionVar,
     },
     vdaf::{
         prio2::{
@@ -26,7 +26,7 @@ use crate::{
     VdafConfig,
 };
 use prio::{
-    codec::{encode_u16_items, CodecError, Encode},
+    codec::{CodecError, Encode},
     field::{Field128, Field64, FieldPrio2},
     vdaf::{
         prio2::{Prio2PrepareShare, Prio2PrepareState},
@@ -135,20 +135,23 @@ impl VdafConfig {
     ///
     /// * `measurement` is the measurement.
     //
-    // TODO spec: Decide if truncating the timestamp should be a MAY or SHOULD.
+    // TODO(issue #100): Truncate the timestamp, as required in DAP-02.
     pub fn produce_report(
         &self,
         hpke_config_list: &[HpkeConfig],
-        now: u64,
+        time: Time,
         task_id: &Id,
         measurement: DapMeasurement,
     ) -> Result<Report, DapError> {
         let mut rng = thread_rng();
-        let nonce = Nonce {
-            time: now,
-            rand: rng.gen(),
+        let metadata = ReportMetadata {
+            nonce: Nonce(rng.gen()),
+            time,
+            extensions: Vec::new(),
         };
 
+        // TODO(issue #100) Set this to the output of `shard()`.
+        let public_share = Vec::new();
         let encoded_input_shares = match self {
             Self::Prio3(prio3_config) => prio3_shard(prio3_config, measurement)?,
             Self::Prio2 { dimension } => prio2_shard(*dimension, measurement)?,
@@ -163,10 +166,11 @@ impl VdafConfig {
         info[..N].copy_from_slice(CTX_INPUT_SHARE);
         info[N] = CTX_ROLE_CLIENT; // Sender role (receiver role set below)
 
-        let mut aad = Vec::with_capacity(50);
+        let mut aad = Vec::with_capacity(58);
         task_id.encode(&mut aad);
-        nonce.encode(&mut aad);
-        0_u16.encode(&mut aad); // Empty extensions
+        metadata.encode(&mut aad);
+        // TODO spec: Consider folding the public share into a field called "header".
+        encode_prefixed_bytes::<4>(&mut aad, &public_share);
 
         let mut encrypted_input_shares = Vec::with_capacity(encoded_input_shares.len());
         for (i, (hpke_config, input_share_data)) in hpke_config_list
@@ -190,8 +194,8 @@ impl VdafConfig {
 
         Ok(Report {
             task_id: task_id.clone(),
-            nonce,
-            extensions: vec![],
+            metadata,
+            public_share,
             encrypted_input_shares,
         })
     }
@@ -217,8 +221,8 @@ impl VdafConfig {
         is_leader: bool,
         verify_key: &VdafVerifyKey,
         task_id: &Id,
-        nonce: &Nonce,
-        extensions: &[Extension],
+        metadata: &ReportMetadata,
+        public_share: &[u8],
         encrypted_input_share: &HpkeCiphertext,
     ) -> Result<(VdafState, VdafMessage), DapError> {
         const N: usize = CTX_INPUT_SHARE.len();
@@ -231,18 +235,16 @@ impl VdafConfig {
             CTX_ROLE_HELPER
         }; // Receiver role
 
-        let mut aad = Vec::with_capacity(50 + extensions.len());
+        let mut aad = Vec::with_capacity(58);
         task_id.encode(&mut aad);
-        let nonce_start = aad.len();
-        nonce.encode(&mut aad);
-        let nonce_end = aad.len();
-        encode_u16_items(&mut aad, &(), extensions);
+        metadata.encode(&mut aad);
+        // TODO spec: Consider folding the public share into a field called "header".
+        encode_prefixed_bytes::<4>(&mut aad, public_share);
 
         let input_share_data = decrypter
             .hpke_decrypt(task_id, &info, &aad, encrypted_input_share)
             .await?;
 
-        let nonce_data = &aad[nonce_start..nonce_end];
         let agg_id = if is_leader { 0 } else { 1 };
         match (self, verify_key) {
             (Self::Prio3(ref prio3_config), VdafVerifyKey::Prio3(ref verify_key)) => {
@@ -250,7 +252,7 @@ impl VdafConfig {
                     prio3_config,
                     verify_key,
                     agg_id,
-                    nonce_data,
+                    metadata.nonce.as_ref(),
                     &input_share_data,
                 )?)
             }
@@ -259,7 +261,7 @@ impl VdafConfig {
                     *dimension,
                     verify_key,
                     agg_id,
-                    nonce_data,
+                    metadata.nonce.as_ref(),
                     &input_share_data,
                 )?)
             }
@@ -295,13 +297,13 @@ impl VdafConfig {
         let mut states = Vec::with_capacity(reports.len());
         let mut seq = Vec::with_capacity(reports.len());
         for report in reports.into_iter() {
-            if processed.contains(&report.nonce) {
+            if processed.contains(&report.metadata.nonce) {
                 return Err(DapError::fatal(
                     "tried to process report sequence with non-unique nonces",
                 )
                 .into());
             }
-            processed.insert(report.nonce.clone());
+            processed.insert(report.metadata.nonce.clone());
 
             if &report.task_id != task_id || report.encrypted_input_shares.len() != 2 {
                 return Err(
@@ -320,17 +322,22 @@ impl VdafConfig {
                     true, // is_leader
                     verify_key,
                     task_id,
-                    &report.nonce,
-                    &report.extensions,
+                    &report.metadata,
+                    &report.public_share,
                     &leader_share,
                 )
                 .await
             {
                 Ok((step, message)) => {
-                    states.push((step, message, report.nonce.clone()));
+                    states.push((
+                        step,
+                        message,
+                        report.metadata.time,
+                        report.metadata.nonce.clone(),
+                    ));
                     seq.push(ReportShare {
-                        nonce: report.nonce,
-                        extensions: report.extensions,
+                        metadata: report.metadata,
+                        public_share: report.public_share,
                         encrypted_input_share: helper_share,
                     });
                 }
@@ -354,6 +361,7 @@ impl VdafConfig {
                 task_id: task_id.clone(),
                 agg_job_id: agg_job_id.clone(),
                 agg_param: Vec::default(),
+                batch_param: BatchParameter::TimeInterval,
                 report_shares: seq,
             },
         ))
@@ -393,10 +401,10 @@ impl VdafConfig {
         let mut states = Vec::with_capacity(num_reports);
         let mut transitions = Vec::with_capacity(num_reports);
         for report_share in agg_init_req.report_shares.iter() {
-            if processed.contains(&report_share.nonce) {
+            if processed.contains(&report_share.metadata.nonce) {
                 return Err(DapAbort::UnrecognizedMessage);
             }
-            processed.insert(report_share.nonce.clone());
+            processed.insert(report_share.metadata.nonce.clone());
 
             let var = match self
                 .consume_report_share(
@@ -404,8 +412,8 @@ impl VdafConfig {
                     false, // is_leader
                     verify_key,
                     &agg_init_req.task_id,
-                    &report_share.nonce,
-                    &report_share.extensions,
+                    &report_share.metadata,
+                    &report_share.public_share,
                     &report_share.encrypted_input_share,
                 )
                 .await
@@ -415,7 +423,11 @@ impl VdafConfig {
                         Self::Prio3(..) => prio3_encode_prepare_message(&message),
                         Self::Prio2 { .. } => prio2_encode_prepare_message(&message),
                     };
-                    states.push((step, report_share.nonce.clone()));
+                    states.push((
+                        step,
+                        report_share.metadata.time,
+                        report_share.metadata.nonce.clone(),
+                    ));
                     TransitionVar::Continued(message_data)
                 }
 
@@ -425,7 +437,7 @@ impl VdafConfig {
             };
 
             transitions.push(Transition {
-                nonce: report_share.nonce.clone(),
+                nonce: report_share.metadata.nonce.clone(),
                 var,
             });
         }
@@ -461,7 +473,7 @@ impl VdafConfig {
 
         let mut seq = Vec::with_capacity(state.seq.len());
         let mut states = Vec::with_capacity(state.seq.len());
-        for (helper, (leader_step, leader_message, leader_nonce)) in
+        for (helper, (leader_step, leader_message, leader_time, leader_nonce)) in
             agg_resp.transitions.into_iter().zip(state.seq.into_iter())
         {
             // TODO spec: Consider removing the nonce from the AggregateResp.
@@ -503,7 +515,7 @@ impl VdafConfig {
 
                     states.push((
                         DapOutputShare {
-                            time: leader_nonce.time,
+                            time: leader_time,
                             checksum: checksum.as_ref().try_into().unwrap(),
                             data,
                         },
@@ -554,7 +566,7 @@ impl VdafConfig {
     ) -> Result<DapHelperTransition<AggregateResp>, DapAbort> {
         let mut processed = HashSet::with_capacity(state.seq.len());
         let mut recognized = HashSet::with_capacity(state.seq.len());
-        for (_, nonce) in state.seq.iter() {
+        for (_, _, nonce) in state.seq.iter() {
             recognized.insert(nonce.clone());
         }
 
@@ -576,7 +588,7 @@ impl VdafConfig {
                 return Err(DapAbort::UnrecognizedMessage);
             }
 
-            for (helper_step, helper_nonce) in &mut helper_iter {
+            for (helper_step, helper_time, helper_nonce) in &mut helper_iter {
                 processed.insert(helper_nonce.clone());
                 if helper_nonce != leader.nonce {
                     // Presumably the leader has skipped this report.
@@ -607,7 +619,7 @@ impl VdafConfig {
                         );
 
                         out_shares.push(DapOutputShare {
-                            time: helper_nonce.time,
+                            time: helper_time,
                             checksum: checksum.as_ref().try_into().unwrap(),
                             data,
                         });
@@ -699,10 +711,10 @@ impl VdafConfig {
         &self,
         hpke_config: &HpkeConfig,
         task_id: &Id,
-        batch_interval: &Interval,
+        batch_selector: &BatchSelector,
         agg_share: &DapAggregateShare,
     ) -> Result<HpkeCiphertext, DapAbort> {
-        produce_encrypted_agg_share(true, hpke_config, task_id, batch_interval, agg_share)
+        produce_encrypted_agg_share(true, hpke_config, task_id, batch_selector, agg_share)
     }
 
     /// Like [`produce_leader_encrypted_agg_share`] but run by the Helper in response to an
@@ -711,10 +723,10 @@ impl VdafConfig {
         &self,
         hpke_config: &HpkeConfig,
         task_id: &Id,
-        batch_interval: &Interval,
+        batch_selector: &BatchSelector,
         agg_share: &DapAggregateShare,
     ) -> Result<HpkeCiphertext, DapAbort> {
-        produce_encrypted_agg_share(false, hpke_config, task_id, batch_interval, agg_share)
+        produce_encrypted_agg_share(false, hpke_config, task_id, batch_selector, agg_share)
     }
 
     /// Decrypt and unshard a sequence of aggregate shares. This method is run by the Collector
@@ -736,7 +748,7 @@ impl VdafConfig {
         &self,
         decrypter: &impl HpkeDecrypter<'_>,
         task_id: &Id,
-        batch_interval: &Interval,
+        batch_selector: &BatchSelector,
         encrypted_agg_shares: Vec<HpkeCiphertext>,
     ) -> Result<DapAggregateResult, DapError> {
         const N: usize = CTX_AGG_SHARE.len();
@@ -746,7 +758,7 @@ impl VdafConfig {
 
         let mut aad = Vec::with_capacity(40);
         task_id.encode(&mut aad);
-        batch_interval.encode(&mut aad);
+        batch_selector.encode(&mut aad);
 
         let mut agg_shares = Vec::with_capacity(encrypted_agg_shares.len());
         for (i, agg_share_ciphertext) in encrypted_agg_shares.iter().enumerate() {
@@ -779,7 +791,7 @@ fn produce_encrypted_agg_share(
     is_leader: bool,
     hpke_config: &HpkeConfig,
     task_id: &Id,
-    batch_interval: &Interval,
+    batch_selector: &BatchSelector,
     agg_share: &DapAggregateShare,
 ) -> Result<HpkeCiphertext, DapAbort> {
     let agg_share_data = agg_share
@@ -801,7 +813,7 @@ fn produce_encrypted_agg_share(
     // TODO spec: Consider adding agg param to AAD.
     let mut aad = Vec::with_capacity(40);
     task_id.encode(&mut aad);
-    batch_interval.encode(&mut aad);
+    batch_selector.encode(&mut aad);
 
     let (enc, payload) = hpke_config
         .encrypt(&info, &aad, &agg_share_data)


Test patch:
diff --git a/daphne/src/messages_test.rs b/daphne/src/messages_test.rs
index cee9445..cacf3c6 100644
--- a/daphne/src/messages_test.rs
+++ b/daphne/src/messages_test.rs
@@ -2,23 +2,12 @@
 // SPDX-License-Identifier: BSD-3-Clause
 
 use crate::messages::{
-    AggregateContinueReq, AggregateInitializeReq, AggregateResp, Extension, HpkeAeadId,
-    HpkeCiphertext, HpkeConfig, HpkeKdfId, HpkeKemId, Id, Nonce, Report, ReportShare, Transition,
-    TransitionVar,
+    AggregateContinueReq, AggregateInitializeReq, AggregateResp, BatchParameter, Extension,
+    HpkeAeadId, HpkeCiphertext, HpkeConfig, HpkeKdfId, HpkeKemId, Id, Nonce, Report,
+    ReportMetadata, ReportShare, Transition, TransitionVar,
 };
 use prio::codec::{Decode, Encode};
 
-#[test]
-fn read_nonce() {
-    let data = [
-        0, 0, 0, 0, 97, 152, 50, 20, 23, 23, 23, 23, 23, 23, 23, 23, 23, 23, 23, 23, 23, 23, 23, 23,
-    ];
-
-    let nonce = Nonce::get_decoded(&data).unwrap();
-    assert_eq!(nonce.time, 1637364244);
-    assert_eq!(nonce.rand, [23; 16]);
-}
-
 #[test]
 fn read_report() {
     let report = Report {
@@ -26,14 +15,15 @@ fn read_report() {
             1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10,
             11, 12, 13, 14, 15, 16,
         ]),
-        nonce: Nonce {
+        metadata: ReportMetadata {
             time: 1637364244,
-            rand: [23; 16],
+            nonce: Nonce([23; 16]),
+            extensions: vec![Extension::Unhandled {
+                typ: 0xfff,
+                payload: b"some extension".to_vec(),
+            }],
         },
-        extensions: vec![Extension::Unhandled {
-            typ: 0xfff,
-            payload: b"some extension".to_vec(),
-        }],
+        public_share: b"public share".to_vec(),
         encrypted_input_shares: vec![
             HpkeCiphertext {
                 config_id: 23,
@@ -57,13 +47,17 @@ fn read_agg_init_req() {
         task_id: Id([23; 32]),
         agg_job_id: Id([1; 32]),
         agg_param: b"this is an aggregation parameter".to_vec(),
+        batch_param: BatchParameter::FixedSize {
+            batch_id: Id([0; 32]),
+        },
         report_shares: vec![
             ReportShare {
-                nonce: Nonce {
+                metadata: ReportMetadata {
                     time: 1637361337,
-                    rand: [99; 16],
+                    nonce: Nonce([99; 16]),
+                    extensions: Vec::default(),
                 },
-                extensions: Vec::default(),
+                public_share: b"public share".to_vec(),
                 encrypted_input_share: HpkeCiphertext {
                     config_id: 23,
                     enc: b"encapsulated key".to_vec(),
@@ -71,11 +65,12 @@ fn read_agg_init_req() {
                 },
             },
             ReportShare {
-                nonce: Nonce {
+                metadata: ReportMetadata {
                     time: 163736423,
-                    rand: [17; 16],
+                    nonce: Nonce([17; 16]),
+                    extensions: Vec::default(),
                 },
-                extensions: Vec::default(),
+                public_share: b"public share".to_vec(),
                 encrypted_input_share: HpkeCiphertext {
                     config_id: 0,
                     enc: vec![],
@@ -96,17 +91,11 @@ fn read_agg_cont_req() {
         agg_job_id: Id([1; 32]),
         transitions: vec![
             Transition {
-                nonce: Nonce {
-                    time: 1637361337,
-                    rand: [0; 16],
-                },
+                nonce: Nonce([0; 16]),
                 var: TransitionVar::Continued(b"this is a VDAF-specific message".to_vec()),
             },
             Transition {
-                nonce: Nonce {
-                    time: 163736423,
-                    rand: [1; 16],
-                },
+                nonce: Nonce([1; 16]),
                 var: TransitionVar::Continued(
                     b"believe it or not this is *also* a VDAF-specific message".to_vec(),
                 ),
@@ -123,17 +112,11 @@ fn read_agg_resp() {
     let want = AggregateResp {
         transitions: vec![
             Transition {
-                nonce: Nonce {
-                    time: 1637361337,
-                    rand: [22; 16],
-                },
+                nonce: Nonce([22; 16]),
                 var: TransitionVar::Continued(b"this is a VDAF-specific message".to_vec()),
             },
             Transition {
-                nonce: Nonce {
-                    time: 163736423,
-                    rand: [255; 16],
-                },
+                nonce: Nonce([255; 16]),
                 var: TransitionVar::Continued(
                     b"believe it or not this is *also* a VDAF-specific message".to_vec(),
                 ),
diff --git a/daphne/src/roles_test.rs b/daphne/src/roles_test.rs
index 1efcb70..f4dbe55 100644
--- a/daphne/src/roles_test.rs
+++ b/daphne/src/roles_test.rs
@@ -10,8 +10,9 @@ use crate::{
     hpke::HpkeReceiverConfig,
     messages::{
         AggregateContinueReq, AggregateInitializeReq, AggregateResp, AggregateShareReq,
-        AggregateShareResp, CollectReq, CollectResp, HpkeCiphertext, Id, Interval, Nonce, Report,
-        ReportShare, Transition, TransitionFailure, TransitionVar,
+        AggregateShareResp, BatchParameter, BatchSelector, CollectReq, CollectResp, HpkeCiphertext,
+        Id, Interval, Nonce, Query, Report, ReportMetadata, ReportShare, Transition,
+        TransitionFailure, TransitionVar,
     },
     roles::{DapAggregator, DapHelper, DapLeader},
     testing::{
@@ -61,6 +62,7 @@ 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;
 
         DapRequest {
             version,
@@ -69,6 +71,7 @@ impl MockAggregator {
                 task_id: task_id.clone(),
                 agg_job_id: Id(rng.gen()),
                 agg_param: Vec::default(),
+                batch_param,
                 report_shares,
             }
             .get_encoded(),
@@ -114,7 +117,7 @@ impl MockAggregator {
             media_type: Some(MEDIA_TYPE_AGG_SHARE_REQ),
             payload: AggregateShareReq {
                 task_id: task_id.clone(),
-                batch_interval: Interval::default(),
+                batch_selector: BatchSelector::default(),
                 agg_param: Vec::default(),
                 report_count,
                 checksum,
@@ -135,7 +138,7 @@ impl MockAggregator {
             media_type: Some(MEDIA_TYPE_COLLECT_REQ),
             payload: CollectReq {
                 task_id: task_id.clone(),
-                batch_interval: Interval::default(),
+                query: Query::default(),
                 agg_param: Vec::default(),
             }
             .get_encoded(),
@@ -244,7 +247,7 @@ impl MockAggregator {
     ) {
         // Leader: Get Leader's encrypted aggregate share.
         let leader_agg_share = self
-            .get_agg_share(&collect_req.task_id, &collect_req.batch_interval)
+            .get_agg_share(&collect_req.task_id, &collect_req.query)
             .await
             .unwrap();
 
@@ -253,7 +256,7 @@ impl MockAggregator {
             .produce_leader_encrypted_agg_share(
                 &task_config.collector_hpke_config,
                 &collect_req.task_id,
-                &collect_req.batch_interval,
+                &collect_req.query,
                 &leader_agg_share,
             )
             .unwrap();
@@ -261,7 +264,7 @@ impl MockAggregator {
         // Leader: Prepare AggregateShareReq.
         let agg_share_req = AggregateShareReq {
             task_id: collect_req.task_id.clone(),
-            batch_interval: collect_req.batch_interval.clone(),
+            batch_selector: collect_req.query.clone(),
             agg_param: collect_req.agg_param.clone(),
             report_count: leader_agg_share.report_count,
             checksum: leader_agg_share.checksum,
@@ -282,6 +285,7 @@ impl MockAggregator {
 
         // Leader: Complete the collect job by storing CollectResp in LeaderStore.processed.
         let collect_resp = CollectResp {
+            report_count: leader_agg_share.report_count,
             encrypted_agg_shares: vec![leader_enc_agg_share, helper_enc_agg_share],
         };
 
@@ -290,7 +294,7 @@ impl MockAggregator {
             .unwrap();
 
         // Leader: Mark the reports as collected.
-        self.mark_collected(task_id, &agg_share_req.batch_interval)
+        self.mark_collected(task_id, &agg_share_req.batch_selector)
             .await
             .unwrap();
     }
@@ -377,11 +381,12 @@ async fn http_post_collect_unauthorized_request() {
 async fn http_post_aggregate_failure_hpke_decrypt_error() {
     let helper = MockAggregator::new();
     let report_shares = vec![ReportShare {
-        nonce: Nonce {
+        metadata: ReportMetadata {
             time: helper.now,
-            rand: [1; 16],
+            nonce: Nonce([1; 16]),
+            extensions: Vec::default(),
         },
-        extensions: Vec::default(),
+        public_share: b"public share".to_vec(),
         encrypted_input_share: HpkeCiphertext {
             config_id: 23,
             enc: b"invalid encapsulated key".to_vec(),
@@ -412,8 +417,8 @@ async fn http_post_aggregate_transition_continue() {
 
     let report = helper.gen_test_report(task_id);
     let report_shares = vec![ReportShare {
-        nonce: report.nonce,
-        extensions: report.extensions,
+        metadata: report.metadata.clone(),
+        public_share: report.public_share,
         // 1st share is for Leader and the rest is for Helpers (note that there is only 1 helper).
         encrypted_input_share: report.encrypted_input_shares[1].clone(),
     }];
@@ -439,8 +444,8 @@ async fn http_post_aggregate_failure_report_replayed() {
 
     let report = helper.gen_test_report(task_id);
     let report_shares = vec![ReportShare {
-        nonce: report.nonce.clone(),
-        extensions: report.extensions,
+        metadata: report.metadata.clone(),
+        public_share: report.public_share,
         // 1st share is for Leader and the rest is for Helpers (note that there is only 1 helper).
         encrypted_input_share: report.encrypted_input_shares[1].clone(),
     }];
@@ -454,7 +459,7 @@ async fn http_post_aggregate_failure_report_replayed() {
         let mut report_store_mutex_guard = helper.report_store.lock().expect("lock() failed");
         let report_store = report_store_mutex_guard.deref_mut();
         let mut processed = HashSet::new();
-        processed.insert(report.nonce.clone());
+        processed.insert(report.metadata.nonce.clone());
         report_store.insert(
             task_id.clone(),
             ReportStore {
@@ -485,8 +490,8 @@ async fn http_post_aggregate_failure_batch_collected() {
 
     let report = helper.gen_test_report(task_id);
     let report_shares = vec![ReportShare {
-        nonce: report.nonce.clone(),
-        extensions: report.extensions,
+        metadata: report.metadata.clone(),
+        public_share: report.public_share,
         // 1st share is for Leader and the rest is for Helpers (note that there is only 1 helper).
         encrypted_input_share: report.encrypted_input_shares[1].clone(),
     }];
@@ -499,7 +504,7 @@ async fn http_post_aggregate_failure_batch_collected() {
     {
         let mut agg_store_mutex_guard = helper.agg_store.lock().expect("lock() failed");
         let agg_store = agg_store_mutex_guard.deref_mut();
-        let bucket_info = BucketInfo::new(task_config, task_id, &report.nonce);
+        let bucket_info = BucketInfo::new(task_config, task_id, report.metadata.time);
         let agg_store_state = AggStoreState {
             agg_share: DapAggregateShare::default(),
             collected: true,
@@ -527,8 +532,8 @@ async fn http_post_aggregate_abort_helper_state_overwritten() {
 
     let report = helper.gen_test_report(task_id);
     let report_shares = vec![ReportShare {
-        nonce: report.nonce.clone(),
-        extensions: report.extensions,
+        metadata: report.metadata.clone(),
+        public_share: report.public_share,
         // 1st share is for Leader and the rest is for Helpers (note that there is only 1 helper).
         encrypted_input_share: report.encrypted_input_shares[1].clone(),
     }];
@@ -660,7 +665,7 @@ async fn poll_collect_job_test_results() {
     // Collector: Create a CollectReq.
     let collector_collect_req = CollectReq {
         task_id: task_id.clone(),
-        batch_interval: task_config.current_batch_window(leader.now),
+        query: task_config.query_for_current_batch_window(leader.now),
         agg_param: Vec::default(),
     };
     let version = task_config.version.clone();
@@ -688,6 +693,7 @@ async fn poll_collect_job_test_results() {
     let resp = leader.get_pending_collect_jobs().await.unwrap();
     let (collect_id, _collect_req) = &resp[0];
     let collect_resp = CollectResp {
+        report_count: 0,
         encrypted_agg_shares: Vec::default(),
     };
 
@@ -720,9 +726,11 @@ async fn http_post_collect_fail_invalid_batch_interval() {
     // Collector: Create a CollectReq with a very large batch interval.
     let collector_collect_req = CollectReq {
         task_id: task_id.clone(),
-        batch_interval: Interval {
-            start: now - (now % task_config.min_batch_duration),
-            duration: leader.global_config.max_batch_duration + task_config.min_batch_duration,
+        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,
+            },
         },
         agg_param: Vec::default(),
     };
@@ -744,12 +752,14 @@ async fn http_post_collect_fail_invalid_batch_interval() {
     // Collector: Create a CollectReq with a batch interval in the past.
     let collector_collect_req = CollectReq {
         task_id: task_id.clone(),
-        batch_interval: Interval {
-            start: now
-                - (now % task_config.min_batch_duration)
-                - leader.global_config.min_batch_interval_start
-                - task_config.min_batch_duration,
-            duration: task_config.min_batch_duration * 2,
+        query: Query::TimeInterval {
+            batch_interval: Interval {
+                start: now
+                    - (now % task_config.min_batch_duration)
+                    - leader.global_config.min_batch_interval_start
+                    - task_config.min_batch_duration,
+                duration: task_config.min_batch_duration * 2,
+            },
         },
         agg_param: Vec::default(),
     };
@@ -771,11 +781,13 @@ async fn http_post_collect_fail_invalid_batch_interval() {
     // Collector: Create a CollectReq with a batch interval in the future.
     let collector_collect_req = CollectReq {
         task_id: task_id.clone(),
-        batch_interval: Interval {
-            start: now - (now % task_config.min_batch_duration)
-                + leader.global_config.max_batch_interval_end
-                - task_config.min_batch_duration,
-            duration: task_config.min_batch_duration * 2,
+        query: Query::TimeInterval {
+            batch_interval: Interval {
+                start: now - (now % task_config.min_batch_duration)
+                    + leader.global_config.max_batch_interval_end
+                    - task_config.min_batch_duration,
+                duration: task_config.min_batch_duration * 2,
+            },
         },
         agg_param: Vec::default(),
     };
@@ -805,11 +817,13 @@ async fn http_post_collect_succeed_max_batch_interval() {
     // Collector: Create a CollectReq with a very large batch interval.
     let collector_collect_req = CollectReq {
         task_id: task_id.clone(),
-        batch_interval: Interval {
-            start: now
-                - (now % task_config.min_batch_duration)
-                - leader.global_config.max_batch_duration / 2,
-            duration: leader.global_config.max_batch_duration,
+        query: Query::TimeInterval {
+            batch_interval: Interval {
+                start: now
+                    - (now % task_config.min_batch_duration)
+                    - leader.global_config.max_batch_duration / 2,
+                duration: leader.global_config.max_batch_duration,
+            },
         },
         agg_param: Vec::default(),
     };
@@ -848,7 +862,7 @@ async fn http_post_collect_fail_overlapping_batch_interval() {
     // Collector: Create first CollectReq.
     let collector_collect_req = CollectReq {
         task_id: task_id.clone(),
-        batch_interval: task_config.current_batch_window(now),
+        query: task_config.query_for_current_batch_window(now),
         agg_param: Vec::default(),
     };
     let version = task_config.version.clone();
@@ -873,7 +887,7 @@ async fn http_post_collect_fail_overlapping_batch_interval() {
     // Collector: Create second CollectReq.
     let collector_collect_req = CollectReq {
         task_id: task_id.clone(),
-        batch_interval: task_config.current_batch_window(now),
+        query: task_config.query_for_current_batch_window(now),
         agg_param: Vec::default(),
     };
     let version = task_config.version.clone();
@@ -902,7 +916,7 @@ async fn http_post_collect_success() {
     // Collector: Create a CollectReq.
     let collector_collect_req = CollectReq {
         task_id: task_id.clone(),
-        batch_interval: task_config.current_batch_window(leader.now),
+        query: task_config.query_for_current_batch_window(leader.now),
         agg_param: Vec::default(),
     };
     let version = task_config.version.clone();
@@ -991,7 +1005,7 @@ async fn e2e() {
     // Collector: Create a CollectReq.
     let collect_req = CollectReq {
         task_id: task_id.clone(),
-        batch_interval: task_config.current_batch_window(leader.now),
+        query: task_config.query_for_current_batch_window(leader.now),
         agg_param: Vec::default(),
     };
     let version = task_config.version.clone();
diff --git a/daphne/src/testing.rs b/daphne/src/testing.rs
index 660cb68..5a3c325 100644
--- a/daphne/src/testing.rs
+++ b/daphne/src/testing.rs
@@ -7,8 +7,8 @@ use crate::{
     auth::{BearerToken, BearerTokenProvider},
     hpke::{HpkeDecrypter, HpkeReceiverConfig},
     messages::{
-        CollectReq, CollectResp, HpkeCiphertext, HpkeConfig, Id, Interval, Nonce, Report,
-        ReportShare, TransitionFailure,
+        BatchSelector, CollectReq, CollectResp, HpkeCiphertext, HpkeConfig, Id, Nonce, Report,
+        ReportShare, Time, TransitionFailure,
     },
     roles::{DapAggregator, DapAuthorizedSender, DapHelper, DapLeader},
     DapAbort, DapAggregateShare, DapCollectJob, DapError, DapGlobalConfig, DapHelperState,
@@ -272,13 +272,14 @@ impl<'a> DapAggregator<'a, BearerToken> for MockAggregator {
     async fn is_batch_overlapping(
         &self,
         task_id: &Id,
-        batch_interval: &Interval,
+        batch_selector: &BatchSelector,
     ) -> Result<bool, DapError> {
         let mut agg_store_mutex_guard = self
             .agg_store
             .lock()
             .map_err(|e| DapError::Fatal(e.to_string()))?;
         let agg_store = agg_store_mutex_guard.deref_mut();
+        let batch_interval = batch_selector.unwrap_interval();
         for (inner_bucket_info, agg_store_state) in agg_store.iter() {
             if task_id == &inner_bucket_info.task_id
                 && batch_interval.start <= inner_bucket_info.window
@@ -335,7 +336,7 @@ impl<'a> DapAggregator<'a, BearerToken> for MockAggregator {
     async fn get_agg_share(
         &self,
         task_id: &Id,
-        batch_interval: &Interval,
+        batch_selector: &BatchSelector,
     ) -> Result<DapAggregateShare, DapError> {
         // Lock agg_store.
         let mut agg_store_mutex_guard = self
@@ -346,6 +347,7 @@ impl<'a> DapAggregator<'a, BearerToken> for MockAggregator {
 
         // Fetch aggregate shares.
         let mut agg_share = DapAggregateShare::default();
+        let batch_interval = batch_selector.unwrap_interval();
         for (inner_bucket_info, agg_store_state) in agg_store.iter() {
             if task_id == &inner_bucket_info.task_id
                 && batch_interval.start <= inner_bucket_info.window
@@ -365,7 +367,7 @@ impl<'a> DapAggregator<'a, BearerToken> for MockAggregator {
     async fn mark_collected(
         &self,
         task_id: &Id,
-        batch_interval: &Interval,
+        batch_selector: &BatchSelector,
     ) -> Result<(), DapError> {
         // Mark aggregate shares as collected.
         let mut agg_store_mutex_guard = self
@@ -373,6 +375,7 @@ impl<'a> DapAggregator<'a, BearerToken> for MockAggregator {
             .lock()
             .map_err(|e| DapError::Fatal(e.to_string()))?;
         let agg_store = agg_store_mutex_guard.deref_mut();
+        let batch_interval = batch_selector.unwrap_interval();
         for (inner_bucket_info, agg_store_state) in agg_store.iter_mut() {
             if task_id == &inner_bucket_info.task_id
                 && batch_interval.start <= inner_bucket_info.window
@@ -417,17 +420,22 @@ impl<'a> DapHelper<'a, BearerToken> for MockAggregator {
             .or_insert_with(ReportStore::new);
 
         for report_share in report_shares.iter() {
-            let bucket_info = BucketInfo::new(task_config, task_id, &report_share.nonce);
+            let bucket_info = BucketInfo::new(task_config, task_id, report_share.metadata.time);
 
             // Check whether Report has been collected or replayed.
-            if let Err(transition_failure) =
-                self.check_report(&bucket_info, &report_share.nonce, report_store, agg_store)
-            {
-                early_fails.insert(report_share.nonce.clone(), transition_failure);
+            if let Err(transition_failure) = self.check_report(
+                &bucket_info,
+                &report_share.metadata.nonce,
+                report_store,
+                agg_store,
+            ) {
+                early_fails.insert(report_share.metadata.nonce.clone(), transition_failure);
             };
 
             // Mark Report processed.
-            report_store.processed.insert(report_share.nonce.clone());
+            report_store
+                .processed
+                .insert(report_share.metadata.nonce.clone());
         }
 
         Ok(early_fails)
@@ -503,7 +511,7 @@ impl<'a> DapLeader<'a, BearerToken> for MockAggregator {
             .get_task_config_for(task_id)
             .await?
             .ok_or_else(|| DapError::fatal("task not found"))?;
-        let bucket_info = BucketInfo::new(task_config, task_id, &report.nonce);
+        let bucket_info = BucketInfo::new(task_config, task_id, report.metadata.time);
 
         // Lock AggStateStore.
         let agg_store_mutex_guard = self
@@ -523,9 +531,12 @@ impl<'a> DapLeader<'a, BearerToken> for MockAggregator {
             .or_insert_with(ReportStore::new);
 
         // Check whether Report has been collected or replayed.
-        if let Err(transition_failure) =
-            self.check_report(&bucket_info, &report.nonce, report_store, agg_store)
-        {
+        if let Err(transition_failure) = self.check_report(
+            &bucket_info,
+            &report.metadata.nonce,
+            report_store,
+            agg_store,
+        ) {
             return Err(DapError::Transition(transition_failure));
         };
 
@@ -717,12 +728,12 @@ pub(crate) struct HelperStateInfo {
 #[derive(Clone, Eq, Hash, PartialEq, Deserialize, Serialize)]
 pub(crate) struct BucketInfo {
     task_id: Id,
-    window: u64,
+    window: Time,
 }
 
 impl BucketInfo {
-    pub(crate) fn new(task_config: &DapTaskConfig, task_id: &Id, nonce: &Nonce) -> Self {
-        let window = nonce.time - (nonce.time % task_config.min_batch_duration);
+    pub(crate) fn new(task_config: &DapTaskConfig, task_id: &Id, time: Time) -> Self {
+        let window = time - (time % task_config.min_batch_duration);
 
         Self {
             task_id: task_id.clone(),
diff --git a/daphne/src/vdaf/mod_test.rs b/daphne/src/vdaf/mod_test.rs
index a304d41..d077e6f 100644
--- a/daphne/src/vdaf/mod_test.rs
+++ b/daphne/src/vdaf/mod_test.rs
@@ -4,8 +4,8 @@
 use crate::{
     hpke::HpkeReceiverConfig,
     messages::{
-        AggregateContinueReq, AggregateInitializeReq, AggregateResp, HpkeAeadId, HpkeCiphertext,
-        HpkeConfig, HpkeKdfId, HpkeKemId, Id, Interval, Nonce, Report, Transition,
+        AggregateContinueReq, AggregateInitializeReq, AggregateResp, BatchSelector, HpkeAeadId,
+        HpkeCiphertext, HpkeConfig, HpkeKdfId, HpkeKemId, Id, Interval, Nonce, Report, Transition,
         TransitionFailure, TransitionVar,
     },
     DapAbort, DapAggregateResult, DapAggregateShare, DapError, DapHelperState, DapHelperTransition,
@@ -82,8 +82,8 @@ async fn roundtrip_report() {
             true, // is_leader
             &t.vdaf_verify_key,
             &t.task_id,
-            &report.nonce,
-            &report.extensions,
+            &report.metadata,
+            &report.public_share,
             &report.encrypted_input_shares[0],
         )
         .await
@@ -95,8 +95,8 @@ async fn roundtrip_report() {
             false, // is_leader
             &t.vdaf_verify_key,
             &t.task_id,
-            &report.nonce,
-            &report.extensions,
+            &report.metadata,
+            &report.public_share,
             &report.encrypted_input_shares[1],
         )
         .await
@@ -181,14 +181,14 @@ async fn agg_init_req() {
     assert_eq!(agg_init_req.agg_param.len(), 0);
     assert_eq!(agg_init_req.report_shares.len(), 3);
     for (report_shares, report) in agg_init_req.report_shares.iter().zip(reports.iter()) {
-        assert_eq!(report_shares.nonce, report.nonce);
+        assert_eq!(report_shares.metadata.nonce, report.metadata.nonce);
     }
 
     let (helper_state, agg_resp) = t.handle_agg_init_req(agg_init_req).await.unwrap_continue();
     assert_eq!(helper_state.seq.len(), 3);
     assert_eq!(agg_resp.transitions.len(), 3);
     for (sub, report) in agg_resp.transitions.iter().zip(reports.iter()) {
-        assert_eq!(sub.nonce, report.nonce);
+        assert_eq!(sub.nonce, report.metadata.nonce);
     }
 }
 
@@ -272,10 +272,7 @@ async fn agg_resp_abort_unrecognized_nonce() {
 
     // Helper sent a transition with an unrecognized Nonce.
     agg_resp.transitions.push(Transition {
-        nonce: Nonce {
-            time: rng.gen(),
-            rand: rng.gen(),
-        },
+        nonce: Nonce(rng.gen()),
         var: TransitionVar::Continued(b"whatever".to_vec()),
     });
 
@@ -388,11 +385,11 @@ async fn agg_cont_req_skip_vdaf_prep_error() {
     assert_eq!(2, agg_resp.transitions.len());
     assert_eq!(
         agg_resp.transitions[0].nonce,
-        agg_init_req.report_shares[0].nonce
+        agg_init_req.report_shares[0].metadata.nonce
     );
     assert_eq!(
         agg_resp.transitions[1].nonce,
-        agg_init_req.report_shares[2].nonce
+        agg_init_req.report_shares[2].metadata.nonce
     );
 }
 
@@ -411,10 +408,7 @@ async fn agg_cont_abort_unrecognized_nonce() {
     agg_cont_req.transitions.insert(
         1,
         Transition {
-            nonce: Nonce {
-                time: rng.gen(),
-                rand: rng.gen(),
-            },
+            nonce: Nonce(rng.gen()),
             var: TransitionVar::Finished, // Expected transition type for Prio3 at this stage
         },
     );
@@ -480,17 +474,19 @@ async fn encrypted_agg_share() {
         data: Some(VdafAggregateShare::Field64(vec![9.into()].into())),
     };
 
-    let batch_interval = Interval {
-        start: 1637359200,
-        duration: 7200,
+    let batch_selector = BatchSelector::TimeInterval {
+        batch_interval: Interval {
+            start: 1637359200,
+            duration: 7200,
+        },
     };
     let leader_encrypted_agg_share =
-        t.produce_leader_encrypted_agg_share(&batch_interval, &leader_agg_share);
+        t.produce_leader_encrypted_agg_share(&batch_selector, &leader_agg_share);
     let helper_encrypted_agg_share =
-        t.produce_helper_encrypted_agg_share(&batch_interval, &helper_agg_share);
+        t.produce_helper_encrypted_agg_share(&batch_selector, &helper_agg_share);
     let agg_res = t
         .consume_encrypted_agg_shares(
-            &batch_interval,
+            &batch_selector,
             vec![leader_encrypted_agg_share, helper_encrypted_agg_share],
         )
         .await;
@@ -618,7 +614,7 @@ impl<'a> Test<'a> {
         for report_share in agg_init_req.report_shares {
             // Make sure the Leader doesn't try to aggregate these reports again.
             self.early_rejects.insert(
-                report_share.nonce.clone(),
+                report_share.metadata.nonce.clone(),
                 TransitionFailure::ReportReplayed,
             );
         }
@@ -680,14 +676,14 @@ impl<'a> Test<'a> {
 
     fn produce_leader_encrypted_agg_share(
         &self,
-        batch_interval: &Interval,
+        batch_selector: &BatchSelector,
         agg_share: &DapAggregateShare,
     ) -> HpkeCiphertext {
         self.vdaf
             .produce_leader_encrypted_agg_share(
                 &self.collector_hpke_config,
                 &self.task_id,
-                batch_interval,
+                batch_selector,
                 agg_share,
             )
             .unwrap()
@@ -695,14 +691,14 @@ impl<'a> Test<'a> {
 
     fn produce_helper_encrypted_agg_share(
         &self,
-        batch_interval: &Interval,
+        batch_selector: &BatchSelector,
         agg_share: &DapAggregateShare,
     ) -> HpkeCiphertext {
         self.vdaf
             .produce_helper_encrypted_agg_share(
                 &self.collector_hpke_config,
                 &self.task_id,
-                batch_interval,
+                batch_selector,
                 agg_share,
             )
             .unwrap()
@@ -710,14 +706,14 @@ impl<'a> Test<'a> {
 
     async fn consume_encrypted_agg_shares(
         &self,
-        batch_interval: &Interval,
+        batch_selector: &BatchSelector,
         enc_agg_shares: Vec<HpkeCiphertext>,
     ) -> DapAggregateResult {
         self.vdaf
             .consume_encrypted_agg_shares(
                 &self.collector_hpke_receiver_config,
                 &self.task_id,
-                batch_interval,
+                batch_selector,
                 enc_agg_shares,
             )
             .await
@@ -728,9 +724,11 @@ impl<'a> Test<'a> {
         &mut self,
         measurements: Vec<DapMeasurement>,
     ) -> DapAggregateResult {
-        let batch_interval = Interval {
-            start: self.now,
-            duration: 3600,
+        let batch_selector = BatchSelector::TimeInterval {
+            batch_interval: Interval {
+                start: self.now,
+                duration: 3600,
+            },
         };
 
         // Clients: Shard
@@ -762,7 +760,7 @@ impl<'a> Test<'a> {
         assert_eq!(leader_agg_shares.len(), 1);
         let (_batch_window, leader_agg_share) = leader_agg_shares.into_iter().next().unwrap();
         let leader_encrypted_agg_share =
-            self.produce_leader_encrypted_agg_share(&batch_interval, &leader_agg_share);
+            self.produce_leader_encrypted_agg_share(&batch_selector, &leader_agg_share);
 
         // Helper: Aggregation
         let helper_agg_shares =
@@ -770,11 +768,11 @@ impl<'a> Test<'a> {
         assert_eq!(helper_agg_shares.len(), 1);
         let (_batch_window, helper_agg_share) = helper_agg_shares.into_iter().next().unwrap();
         let helper_encrypted_agg_share =
-            self.produce_helper_encrypted_agg_share(&batch_interval, &helper_agg_share);
+            self.produce_helper_encrypted_agg_share(&batch_selector, &helper_agg_share);
 
         // Collector: Unshard
         self.consume_encrypted_agg_shares(
-            &batch_interval,
+            &batch_selector,
             vec![leader_encrypted_agg_share, helper_encrypted_agg_share],
         )
         .await