Repository: cloudflare/daphne (rust)

Issue:
Support the next interop target
As described [here](https://mailarchive.ietf.org/arch/msg/ppm/Q-uN4IAfekbciRDgyHt3tlwYSAA/), the next set of drafts to implement are (DAP-12):
* [draft-ietf-ppm-dap-12](https://datatracker.ietf.org/doc/draft-ietf-ppm-dap/)
* [draft-irtf-ppm-dap-taskprov-01](https://datatracker.ietf.org/doc/draft-ietf-ppm-dap-taskprov/) (WIP as of 2024/10/16)
* [draft-irtf-cfrg-vdaf-12](https://datatracker.ietf.org/doc/draft-irtf-cfrg-vdaf/) for the Prio3
* We also might need to bump [draft-dcook-ppm-dap-interop-test-design](https://datatracker.ietf.org/doc/draft-dcook-ppm-dap-interop-test-design/), but this needs discussion.

The draft set we currently support is (DAP-09):
* draft-ietf-ppm-dap-09
* draft-irtf-cfrg-vdaf-08 for Prio3
* draft-wang-ppm-dap-taskprov-06
* draft-dcook-ppm-dap-interop-test-design-07

We will need to add support for DAP-12 alongside DAP-09.

# Requirements

1. The version of a VDAF, of taskprov, or of the interop API that is used is determined by the version of DAP.  In particular, we won't try to support multiple versions of a VDAF or of the taskprov extension with the same DAP version.
1. Our implementations of Prio2 and Prio3 come from the [prio](https://github.com/divviup/libprio-rs) crate. Version 0.16 is compatible with DAP-09; a future version of the crate, perhaps 0.17, [will be compatible with DAP-12](https://github.com/divviup/libprio-rs/issues/1122). We will need to import both versions of the crate at the same time.
1. DAP-09 MUST support all VDAFs currently supported by the taskprov extension. This includes Prio2, PINE (see below), and some non-standard Prio3SumVec variants.
1. DAP-12 MUST support all currently supported VDAFs, excluding PINE: [draft-chen-cfrg-vdaf-pine-01](https://datatracker.ietf.org/doc/draft-chen-cfrg-vdaf-pine/) is not [API compatible with DAP-12](https://github.com/junyechen1996/draft-chen-cfrg-vdaf-pine/issues/106), so for now we'll only maintain support for DAP-09.

# Notes

1. `DapVersion::Latest` is currently set to DAP-10; this should be replaced with DAP-12.
1. The most important changes since DAP-09 are summarized in the draft's [change log](https://www.ietf.org/archive/id/draft-ietf-ppm-dap-12.html#section-1.1). Most are relatively minor, but it will be necessary to plumb the DAP version down more code paths to, for example, account for changes in message formats.
1. The most significant change since DAP-09 is asynchronous aggregation. This feature is _optional_, however, and we'll handle it in a separate issue. See #697.
1. Note that there are some known discrepancies between our implementation and the spec: see #522 and #557.

# Validation

1. This repo has unit tests and end-to-end tests, but for the most part they don't enforce interop. We mainly rely on additional tests in the internal repo for the Helper deployment. These will be useful for making sure we don't regress DAP-09.
1. [Janus is currently working on DAP-12 support as well](https://github.com/divviup/janus/issues/3436). The `interop/` folder in this directory defines a docker image they can use to automatically run a bunch of interop tests with our Helper; we can use this to test interop with DAP-12.


Pull request:
DAP-13: Rename `Transition`
Closes #698.

* Rename `Transition` to `PrepareResp`


Patch:
diff --git a/crates/daphne/src/messages/mod.rs b/crates/daphne/src/messages/mod.rs
index d0674e3..462f34b 100644
--- a/crates/daphne/src/messages/mod.rs
+++ b/crates/daphne/src/messages/mod.rs
@@ -632,16 +632,15 @@ impl ParameterizedDecode<DapVersion> for AggregationJobInitReq {
     }
 }
 
-/// Transition message. This conveyes a message sent from one Aggregator to another during the
-/// preparation phase of VDAF evaluation.
+/// The `PrepareResp` message.
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(any(test, feature = "test-utils"), derive(deepsize::DeepSizeOf))]
-pub struct Transition {
+pub struct PrepareResp {
     pub report_id: ReportId,
-    pub var: TransitionVar,
+    pub var: PrepareRespVar,
 }
 
-impl ParameterizedEncode<DapVersion> for Transition {
+impl ParameterizedEncode<DapVersion> for PrepareResp {
     fn encode_with_param(
         &self,
         version: &DapVersion,
@@ -652,38 +651,38 @@ impl ParameterizedEncode<DapVersion> for Transition {
         Ok(())
     }
 }
-impl ParameterizedDecode<DapVersion> for Transition {
+impl ParameterizedDecode<DapVersion> for PrepareResp {
     fn decode_with_param(
         version: &DapVersion,
         bytes: &mut Cursor<&[u8]>,
     ) -> Result<Self, CodecError> {
         Ok(Self {
             report_id: ReportId::decode(bytes)?,
-            var: TransitionVar::decode_with_param(version, bytes)?,
+            var: PrepareRespVar::decode_with_param(version, bytes)?,
         })
     }
 }
 
-/// Transition message variant.
+/// A variant of `PrepareResp`.
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(any(test, feature = "test-utils"), derive(deepsize::DeepSizeOf))]
-pub enum TransitionVar {
-    Continued(Vec<u8>),
-    Failed(ReportError),
+pub enum PrepareRespVar {
+    Continue(Vec<u8>),
+    Reject(ReportError),
 }
 
-impl ParameterizedEncode<DapVersion> for TransitionVar {
+impl ParameterizedEncode<DapVersion> for PrepareRespVar {
     fn encode_with_param(
         &self,
         version: &DapVersion,
         bytes: &mut Vec<u8>,
     ) -> Result<(), CodecError> {
         match self {
-            TransitionVar::Continued(vdaf_message) => {
+            PrepareRespVar::Continue(vdaf_message) => {
                 0_u8.encode(bytes)?;
                 encode_u32_bytes(bytes, vdaf_message)?;
             }
-            TransitionVar::Failed(err) => {
+            PrepareRespVar::Reject(err) => {
                 2_u8.encode(bytes)?;
                 err.encode_with_param(version, bytes)?;
             }
@@ -692,22 +691,23 @@ impl ParameterizedEncode<DapVersion> for TransitionVar {
     }
 }
 
-impl ParameterizedDecode<DapVersion> for TransitionVar {
+impl ParameterizedDecode<DapVersion> for PrepareRespVar {
     fn decode_with_param(
         version: &DapVersion,
         bytes: &mut Cursor<&[u8]>,
     ) -> Result<Self, CodecError> {
         match u8::decode(bytes)? {
-            0 => Ok(Self::Continued(decode_u32_bytes(bytes)?)),
-            2 => Ok(Self::Failed(ReportError::decode_with_param(
+            0 => Ok(Self::Continue(decode_u32_bytes(bytes)?)),
+            2 => Ok(Self::Reject(ReportError::decode_with_param(
                 version, bytes,
             )?)),
+            // Treat `finished` as an unexpected value, since we don't implement multi-round VDAFs.
             _ => Err(CodecError::UnexpectedValue),
         }
     }
 }
 
-/// Transition error.
+/// Report error.
 #[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq, thiserror::Error, Hash)]
 #[serde(rename_all = "snake_case")]
 #[cfg_attr(any(test, feature = "test-utils"), derive(deepsize::DeepSizeOf))]
@@ -958,7 +958,7 @@ const AGG_JOB_RESP_READY: u8 = 1;
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(any(test, feature = "test-utils"), derive(deepsize::DeepSizeOf))]
 pub enum AggregationJobResp {
-    Ready { transitions: Vec<Transition> },
+    Ready { prep_resps: Vec<PrepareResp> },
     Processing,
 }
 
@@ -967,7 +967,7 @@ impl AggregationJobResp {
     #[track_caller]
     pub fn unwrap_ready(self) -> crate::protocol::ReadyAggregationJobResp {
         match self {
-            Self::Ready { transitions } => crate::protocol::ReadyAggregationJobResp { transitions },
+            Self::Ready { prep_resps } => crate::protocol::ReadyAggregationJobResp { prep_resps },
             Self::Processing => panic!("unwrapped a Processing value"),
         }
     }
@@ -980,12 +980,12 @@ impl ParameterizedEncode<DapVersion> for AggregationJobResp {
         bytes: &mut Vec<u8>,
     ) -> Result<(), CodecError> {
         match (self, version) {
-            (Self::Ready { transitions }, DapVersion::Draft09) => {
-                encode_u32_items(bytes, version, transitions)
+            (Self::Ready { prep_resps }, DapVersion::Draft09) => {
+                encode_u32_items(bytes, version, prep_resps)
             }
-            (Self::Ready { transitions }, DapVersion::Latest) => {
+            (Self::Ready { prep_resps }, DapVersion::Latest) => {
                 AGG_JOB_RESP_READY.encode(bytes)?;
-                encode_u32_items(bytes, version, transitions)
+                encode_u32_items(bytes, version, prep_resps)
             }
             (Self::Processing, DapVersion::Draft09) => Err(CodecError::Other(
                 "draft09: can't represent an agg job resp that's being processed".into(),
@@ -1002,12 +1002,12 @@ impl ParameterizedDecode<DapVersion> for AggregationJobResp {
     ) -> Result<Self, CodecError> {
         match version {
             DapVersion::Draft09 => Ok(Self::Ready {
-                transitions: decode_u32_items(version, bytes)?,
+                prep_resps: decode_u32_items(version, bytes)?,
             }),
             DapVersion::Latest => match u8::decode(bytes)? {
                 AGG_JOB_RESP_PROCESSING => Ok(Self::Processing),
                 AGG_JOB_RESP_READY => Ok(Self::Ready {
-                    transitions: decode_u32_items(version, bytes)?,
+                    prep_resps: decode_u32_items(version, bytes)?,
                 }),
                 _ => Err(CodecError::UnexpectedValue),
             },
@@ -1936,20 +1936,20 @@ mod test {
         ];
 
         let want = AggregationJobResp::Ready {
-            transitions: vec![
-                Transition {
+            prep_resps: vec![
+                PrepareResp {
                     report_id: ReportId([22; 16]),
-                    var: TransitionVar::Continued(b"this is a VDAF-specific message".to_vec()),
+                    var: PrepareRespVar::Continue(b"this is a VDAF-specific message".to_vec()),
                 },
-                Transition {
+                PrepareResp {
                     report_id: ReportId([255; 16]),
-                    var: TransitionVar::Continued(
+                    var: PrepareRespVar::Continue(
                         b"believe it or not this is *also* a VDAF-specific message".to_vec(),
                     ),
                 },
-                Transition {
+                PrepareResp {
                     report_id: ReportId([17; 16]),
-                    var: TransitionVar::Failed(ReportError::TaskExpired),
+                    var: PrepareRespVar::Reject(ReportError::TaskExpired),
                 },
             ],
         };
diff --git a/crates/daphne/src/protocol/aggregator.rs b/crates/daphne/src/protocol/aggregator.rs
index ee5eb85..a2ce008 100644
--- a/crates/daphne/src/protocol/aggregator.rs
+++ b/crates/daphne/src/protocol/aggregator.rs
@@ -13,8 +13,8 @@ use crate::{
     hpke::{info_and_aad, HpkeConfig, HpkeDecrypter},
     messages::{
         self, encode_u32_bytes, AggregationJobInitReq, Base64Encode, BatchSelector, HpkeCiphertext,
-        PartialBatchSelector, PrepareInit, Report, ReportError, ReportId, ReportShare, TaskId,
-        Transition, TransitionVar,
+        PartialBatchSelector, PrepareInit, PrepareResp, PrepareRespVar, Report, ReportError,
+        ReportId, ReportShare, TaskId,
     },
     metrics::{DaphneMetrics, ReportStatus},
     protocol::{decode_ping_pong_framed, PingPongMessageType},
@@ -289,12 +289,12 @@ impl DapTaskConfig {
     ) -> Result<(DapAggregateSpan<DapAggregateShare>, ReadyAggregationJobResp), DapError> {
         let num_reports = initialized_reports.len();
         let mut agg_span = DapAggregateSpan::default();
-        let mut transitions = Vec::with_capacity(num_reports);
+        let mut prep_resps = Vec::with_capacity(num_reports);
 
         for initialized_report in initialized_reports {
             let status = report_status.get(&initialized_report.metadata().id);
             let var = match status {
-                Some(ReportProcessedStatus::Rejected(err)) => TransitionVar::Failed(*err),
+                Some(ReportProcessedStatus::Rejected(err)) => PrepareRespVar::Reject(*err),
                 Some(ReportProcessedStatus::Aggregated) | None => match initialized_report {
                     InitializedReport::Ready {
                         metadata,
@@ -332,7 +332,7 @@ impl DapTaskConfig {
                                 outbound.push(PingPongMessageType::Finish as u8);
                                 encode_u32_bytes(&mut outbound, &prep_msg)
                                     .map_err(DapError::encoding)?;
-                                TransitionVar::Continued(outbound)
+                                PrepareRespVar::Continue(outbound)
                             }
 
                             Err(
@@ -342,7 +342,7 @@ impl DapTaskConfig {
                                 | VdafError::Vdaf(..)),
                             ) => {
                                 tracing::warn!(error = ?e, "rejecting report");
-                                TransitionVar::Failed(ReportError::VdafPrepError)
+                                PrepareRespVar::Reject(ReportError::VdafPrepError)
                             }
 
                             Err(VdafError::Dap(e)) => return Err(e),
@@ -352,17 +352,17 @@ impl DapTaskConfig {
                     InitializedReport::Rejected {
                         metadata: _,
                         report_err,
-                    } => TransitionVar::Failed(*report_err),
+                    } => PrepareRespVar::Reject(*report_err),
                 },
             };
 
-            transitions.push(Transition {
+            prep_resps.push(PrepareResp {
                 report_id: initialized_report.metadata().id,
                 var,
             });
         }
 
-        Ok((agg_span, ReadyAggregationJobResp { transitions }))
+        Ok((agg_span, ReadyAggregationJobResp { prep_resps }))
     }
 
     /// Leader: Consume the `AggregationJobResp` message sent by the Helper and compute the
@@ -374,11 +374,11 @@ impl DapTaskConfig {
         agg_job_resp: ReadyAggregationJobResp,
         metrics: &dyn DaphneMetrics,
     ) -> Result<DapAggregateSpan<DapAggregateShare>, DapError> {
-        if agg_job_resp.transitions.len() != state.seq.len() {
+        if agg_job_resp.prep_resps.len() != state.seq.len() {
             return Err(DapAbort::InvalidMessage {
                 detail: format!(
                     "aggregation job response has {} reports; expected {}",
-                    agg_job_resp.transitions.len(),
+                    agg_job_resp.prep_resps.len(),
                     state.seq.len(),
                 ),
                 task_id: *task_id,
@@ -387,7 +387,7 @@ impl DapTaskConfig {
         }
 
         let mut agg_span = DapAggregateSpan::default();
-        for (helper, leader) in zip(agg_job_resp.transitions, state.seq) {
+        for (helper, leader) in zip(agg_job_resp.prep_resps, state.seq) {
             if helper.report_id != leader.report_id {
                 return Err(DapAbort::InvalidMessage {
                     detail: format!(
@@ -400,7 +400,7 @@ impl DapTaskConfig {
             }
 
             let prep_msg = match &helper.var {
-                TransitionVar::Continued(inbound) => {
+                PrepareRespVar::Continue(ref inbound) => {
                     // Decode the ping-pong "finish" message frame (draft-irtf-cfrg-vdaf-08,
                     // Section 5.8). Abort the aggregation job if not found.
                     let Ok(prep_msg) =
@@ -419,7 +419,7 @@ impl DapTaskConfig {
                 }
 
                 // Skip report that can't be processed any further.
-                TransitionVar::Failed(err) => {
+                PrepareRespVar::Reject(err) => {
                     tracing::warn!("report rejected by Helper: {err:?}");
                     metrics.report_inc_by(ReportStatus::Rejected(*err), 1);
                     continue;
diff --git a/crates/daphne/src/protocol/mod.rs b/crates/daphne/src/protocol/mod.rs
index bbbecde..162947f 100644
--- a/crates/daphne/src/protocol/mod.rs
+++ b/crates/daphne/src/protocol/mod.rs
@@ -64,13 +64,13 @@ fn decode_ping_pong_framed(
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(any(test, feature = "test-utils"), derive(deepsize::DeepSizeOf))]
 pub struct ReadyAggregationJobResp {
-    pub transitions: Vec<messages::Transition>,
+    pub prep_resps: Vec<messages::PrepareResp>,
 }
 
 impl From<ReadyAggregationJobResp> for messages::AggregationJobResp {
     fn from(value: ReadyAggregationJobResp) -> Self {
         Self::Ready {
-            transitions: value.transitions,
+            prep_resps: value.prep_resps,
         }
     }
 }
@@ -84,7 +84,7 @@ mod test {
         hpke::{HpkeAeadId, HpkeConfig, HpkeKdfId, HpkeKemId},
         messages::{
             AggregationJobInitReq, BatchSelector, Extension, Interval, PartialBatchSelector,
-            PrepareInit, Report, ReportError, ReportId, ReportShare, Transition, TransitionVar,
+            PrepareInit, PrepareResp, PrepareRespVar, Report, ReportError, ReportId, ReportShare,
         },
         test_versions,
         testing::AggregationJobTest,
@@ -266,7 +266,7 @@ mod test {
 
         let (agg_span, agg_job_resp) = t.handle_agg_job_req(agg_job_init_req);
         assert_eq!(agg_span.report_count(), 3);
-        assert_eq!(agg_job_resp.unwrap_ready().transitions.len(), 3);
+        assert_eq!(agg_job_resp.unwrap_ready().prep_resps.len(), 3);
     }
 
     test_versions! { produce_agg_job_req }
@@ -393,10 +393,10 @@ mod test {
         let (_agg_span, agg_job_resp) = t.handle_agg_job_req(agg_job_init_req);
 
         let agg_job_resp = agg_job_resp.unwrap_ready();
-        assert_eq!(agg_job_resp.transitions.len(), 1);
+        assert_eq!(agg_job_resp.prep_resps.len(), 1);
         assert_matches!(
-            agg_job_resp.transitions[0].var,
-            TransitionVar::Failed(ReportError::HpkeDecryptError)
+            agg_job_resp.prep_resps[0].var,
+            PrepareRespVar::Reject(ReportError::HpkeDecryptError)
         );
     }
 
@@ -432,10 +432,10 @@ mod test {
         let (_agg_span, agg_job_resp) = t.handle_agg_job_req(agg_job_init_req);
 
         let agg_job_resp = agg_job_resp.unwrap_ready();
-        assert_eq!(agg_job_resp.transitions.len(), 1);
+        assert_eq!(agg_job_resp.prep_resps.len(), 1);
         assert_matches!(
-            agg_job_resp.transitions[0].var,
-            TransitionVar::Failed(ReportError::ReportDropped)
+            agg_job_resp.prep_resps[0].var,
+            PrepareRespVar::Reject(ReportError::ReportDropped)
         );
     }
 
@@ -468,10 +468,10 @@ mod test {
         let (_agg_span, agg_job_resp) = t.handle_agg_job_req(agg_job_init_req);
 
         let agg_job_resp = agg_job_resp.unwrap_ready();
-        assert_eq!(agg_job_resp.transitions.len(), 1);
+        assert_eq!(agg_job_resp.prep_resps.len(), 1);
         assert_matches!(
-            agg_job_resp.transitions[0].var,
-            TransitionVar::Failed(ReportError::ReportTooEarly)
+            agg_job_resp.prep_resps[0].var,
+            PrepareRespVar::Reject(ReportError::ReportTooEarly)
         );
     }
 
@@ -489,10 +489,10 @@ mod test {
         let (_agg_span, agg_job_resp) = t.handle_agg_job_req(agg_job_init_req);
 
         let agg_job_resp = agg_job_resp.unwrap_ready();
-        assert_eq!(agg_job_resp.transitions.len(), 1);
+        assert_eq!(agg_job_resp.prep_resps.len(), 1);
         assert_matches!(
-            agg_job_resp.transitions[0].var,
-            TransitionVar::Failed(ReportError::HpkeUnknownConfigId)
+            agg_job_resp.prep_resps[0].var,
+            PrepareRespVar::Reject(ReportError::HpkeUnknownConfigId)
         );
     }
 
@@ -535,14 +535,14 @@ mod test {
         let (_agg_span, agg_job_resp) = t.handle_agg_job_req(agg_job_init_req);
 
         let agg_job_resp = agg_job_resp.unwrap_ready();
-        assert_eq!(agg_job_resp.transitions.len(), 2);
+        assert_eq!(agg_job_resp.prep_resps.len(), 2);
         assert_matches!(
-            agg_job_resp.transitions[0].var,
-            TransitionVar::Failed(ReportError::VdafPrepError)
+            agg_job_resp.prep_resps[0].var,
+            PrepareRespVar::Reject(ReportError::VdafPrepError)
         );
         assert_matches!(
-            agg_job_resp.transitions[1].var,
-            TransitionVar::Failed(ReportError::VdafPrepError)
+            agg_job_resp.prep_resps[1].var,
+            PrepareRespVar::Reject(ReportError::VdafPrepError)
         );
     }
 
@@ -559,10 +559,10 @@ mod test {
         let (_agg_span, agg_job_resp) = t.handle_agg_job_req(agg_job_init_req);
 
         let mut agg_job_resp = agg_job_resp.unwrap_ready();
-        // Helper sends transitions out of order.
-        let tmp = agg_job_resp.transitions[0].clone();
-        agg_job_resp.transitions[0] = agg_job_resp.transitions[1].clone();
-        agg_job_resp.transitions[1] = tmp;
+        // Helper sends prep_resps out of order.
+        let tmp = agg_job_resp.prep_resps[0].clone();
+        agg_job_resp.prep_resps[0] = agg_job_resp.prep_resps[1].clone();
+        agg_job_resp.prep_resps[1] = tmp;
 
         assert_matches!(
             t.consume_agg_job_resp_expect_err(leader_state, agg_job_resp),
@@ -584,8 +584,8 @@ mod test {
 
         let mut agg_job_resp = agg_job_resp.unwrap_ready();
         // Helper sends a transition twice.
-        let repeated_transition = agg_job_resp.transitions[0].clone();
-        agg_job_resp.transitions.push(repeated_transition);
+        let repeated_transition = agg_job_resp.prep_resps[0].clone();
+        agg_job_resp.prep_resps.push(repeated_transition);
 
         assert_matches!(
             t.consume_agg_job_resp_expect_err(leader_state, agg_job_resp),
@@ -608,9 +608,9 @@ mod test {
 
         let mut agg_job_resp = agg_job_resp.unwrap_ready();
         // Helper sent a transition with an unrecognized report ID.
-        agg_job_resp.transitions.push(Transition {
+        agg_job_resp.prep_resps.push(PrepareResp {
             report_id: ReportId(rng.gen()),
-            var: TransitionVar::Continued(b"whatever".to_vec()),
+            var: PrepareRespVar::Continue(b"whatever".to_vec()),
         });
 
         assert_matches!(
@@ -631,7 +631,7 @@ mod test {
         let mut agg_job_resp = agg_job_resp.unwrap_ready();
         // Helper sent a transition with an unrecognized report ID. Simulate this by flipping the
         // first bit of the report ID.
-        agg_job_resp.transitions[0].report_id.0[0] ^= 1;
+        agg_job_resp.prep_resps[0].report_id.0[0] ^= 1;
 
         assert_matches!(
             t.consume_agg_job_resp_expect_err(leader_state, agg_job_resp),
@@ -712,8 +712,8 @@ mod test {
         let agg_job_resp = agg_job_resp.unwrap_ready();
 
         assert_eq!(2, helper_agg_span.report_count());
-        assert_eq!(3, agg_job_resp.transitions.len());
-        for (transition, prep_init_id) in zip(&agg_job_resp.transitions, prep_init_ids) {
+        assert_eq!(3, agg_job_resp.prep_resps.len());
+        for (transition, prep_init_id) in zip(&agg_job_resp.prep_resps, prep_init_ids) {
             assert_eq!(transition.report_id, prep_init_id);
         }
 
diff --git a/crates/daphne/src/roles/helper/handle_agg_job.rs b/crates/daphne/src/roles/helper/handle_agg_job.rs
index 89d9904..b8f7e1d 100644
--- a/crates/daphne/src/roles/helper/handle_agg_job.rs
+++ b/crates/daphne/src/roles/helper/handle_agg_job.rs
@@ -5,8 +5,8 @@ use super::{check_part_batch, DapHelper, HashedAggregationJobReq};
 use crate::{
     error::DapAbort,
     messages::{
-        AggregationJobInitReq, AggregationJobResp, PartialBatchSelector, ReportError, TaskId,
-        TransitionVar,
+        AggregationJobInitReq, AggregationJobResp, PartialBatchSelector, PrepareRespVar,
+        ReportError, TaskId,
     },
     metrics::ReportStatus,
     protocol::aggregator::ReportProcessedStatus,
@@ -247,7 +247,7 @@ impl HandleAggJob<InitializedReports> {
         // - try to aggregate the output shares into an `DapAggregateShareSpan`
         // - pass it to `try_put_agg_share_span`
         //   - if replays are found, then try again, rejecting the reports that were replayed
-        //   - else break with the finished (of failed) transitions
+        //   - else break with the finished (of failed) prep_resps
         //
         // The reason we do this is because we don't expect replays to happen but we have to guard
         // against them, as such, even though retrying is possibly very expensive, it probably
@@ -311,18 +311,18 @@ impl HandleAggJob<InitializedReports> {
             }
             if !inc_restart_metric.is_completed() {
                 let out_shares_count = agg_job_resp
-                    .transitions
+                    .prep_resps
                     .iter()
-                    .filter(|t| !matches!(t.var, TransitionVar::Failed(..)))
+                    .filter(|t| !matches!(t.var, PrepareRespVar::Reject(..)))
                     .count()
                     .try_into()
                     .expect("usize to fit in u64");
 
                 metrics.report_inc_by(ReportStatus::Aggregated, out_shares_count);
 
-                for transition in &agg_job_resp.transitions {
-                    if let TransitionVar::Failed(failure) = &transition.var {
-                        metrics.report_inc_by(ReportStatus::Rejected(*failure), 1);
+                for transition in &agg_job_resp.prep_resps {
+                    if let PrepareRespVar::Reject(err) = &transition.var {
+                        metrics.report_inc_by(ReportStatus::Rejected(*err), 1);
                     }
                 }
 
@@ -332,7 +332,7 @@ impl HandleAggJob<InitializedReports> {
                 helper.audit_log().on_aggregation_job(
                     &task_id,
                     &task_config,
-                    agg_job_resp.transitions.len() as u64,
+                    agg_job_resp.prep_resps.len() as u64,
                     0, /* vdaf step */
                 );
 
diff --git a/crates/daphne/src/roles/leader/mod.rs b/crates/daphne/src/roles/leader/mod.rs
index efa62bb..a188b65 100644
--- a/crates/daphne/src/roles/leader/mod.rs
+++ b/crates/daphne/src/roles/leader/mod.rs
@@ -374,8 +374,8 @@ async fn run_agg_job<A: DapLeader>(
             .map_err(|e| DapAbort::from_codec_error(e, *task_id))?;
 
     let agg_job_resp = match agg_job_resp {
-        AggregationJobResp::Ready { transitions } => {
-            crate::protocol::ReadyAggregationJobResp { transitions }
+        AggregationJobResp::Ready { prep_resps } => {
+            crate::protocol::ReadyAggregationJobResp { prep_resps }
         }
         AggregationJobResp::Processing => todo!("polling not implemented yet"),
     };
diff --git a/crates/daphne/src/roles/mod.rs b/crates/daphne/src/roles/mod.rs
index a6ecb73..091f46e 100644
--- a/crates/daphne/src/roles/mod.rs
+++ b/crates/daphne/src/roles/mod.rs
@@ -138,8 +138,8 @@ mod test {
         messages::{
             request::RequestBody, AggregateShareReq, AggregationJobId, AggregationJobInitReq,
             AggregationJobResp, BatchId, BatchSelector, Collection, CollectionJobId, CollectionReq,
-            Extension, HpkeCiphertext, Interval, PartialBatchSelector, Query, Report, ReportError,
-            TaskId, Time, TransitionVar,
+            Extension, HpkeCiphertext, Interval, PartialBatchSelector, PrepareRespVar, Query,
+            Report, ReportError, TaskId, Time,
         },
         roles::{helper::HashedAggregationJobReq, leader::WorkItem, DapAggregator},
         testing::InMemoryAggregator,
@@ -627,10 +627,10 @@ mod test {
     //
     //        let resp = t.helper.handle_agg_job_req(&req).await.unwrap();
     //        let agg_job_resp = AggregationJobResp::get_decoded(&resp.payload).unwrap();
-    //        assert_eq!(agg_job_resp.transitions.len(), 1);
+    //        assert_eq!(agg_job_resp.prep_resps.len(), 1);
     //        assert_matches!(
-    //            agg_job_resp.transitions[0].var,
-    //            TransitionVar::Failed(ReportError::TaskExpired)
+    //            agg_job_resp.prep_resps[0].var,
+    //            PrepareRespVar::Reject(ReportError::TaskExpired)
     //        );
     //
     //        assert_eq!(t.helper.audit_log.invocations(), 1);
@@ -751,7 +751,7 @@ mod test {
             .gen_test_agg_job_init_req(task_id, DapAggregationParam::Empty, vec![report])
             .await;
 
-        // Get AggregationJobResp and then extract the transition data from inside.
+        // Get AggregationJobResp and then extract the prep_resp data from inside.
         let agg_job_resp = AggregationJobResp::get_decoded_with_param(
             &version,
             &helper::handle_agg_job_init_req(
@@ -764,12 +764,12 @@ mod test {
             .payload,
         )
         .unwrap();
-        let transition = agg_job_resp.unwrap_ready().transitions.remove(0);
+        let prep_resp = agg_job_resp.unwrap_ready().prep_resps.remove(0);
 
         // Expect failure due to invalid ciphertext.
         assert_matches!(
-            transition.var,
-            TransitionVar::Failed(ReportError::HpkeDecryptError)
+            prep_resp.var,
+            PrepareRespVar::Reject(ReportError::HpkeDecryptError)
         );
     }
 
@@ -856,7 +856,7 @@ mod test {
         _ = leader::handle_upload_req(&*t.leader, req).await;
     }
 
-    async fn handle_agg_job_req_transition_continue(version: DapVersion) {
+    async fn handle_agg_job_req_prep_resp_continue(version: DapVersion) {
         let t = Test::new(version);
         let task_id = &t.time_interval_task_id;
 
@@ -865,7 +865,7 @@ mod test {
             .gen_test_agg_job_init_req(task_id, DapAggregationParam::Empty, vec![report])
             .await;
 
-        // Get AggregationJobResp and then extract the transition data from inside.
+        // Get AggregationJobResp and then extract the prep_resp data from inside.
         let agg_job_resp = AggregationJobResp::get_decoded_with_param(
             &version,
             &helper::handle_agg_job_init_req(
@@ -878,13 +878,13 @@ mod test {
             .payload,
         )
         .unwrap();
-        let transition = agg_job_resp.unwrap_ready().transitions.remove(0);
+        let prep_resp = agg_job_resp.unwrap_ready().prep_resps.remove(0);
 
         // Expect success due to valid ciphertext.
-        assert_matches!(transition.var, TransitionVar::Continued(_));
+        assert_matches!(prep_resp.var, PrepareRespVar::Continue(_));
     }
 
-    async_test_versions! { handle_agg_job_req_transition_continue }
+    async_test_versions! { handle_agg_job_req_prep_resp_continue }
 
     async fn handle_agg_job_req_failure_report_replayed(version: DapVersion) {
         let t = Test::new(version);


Test patch:
diff --git a/crates/daphne/src/testing/mod.rs b/crates/daphne/src/testing/mod.rs
index 7500fad..4eaab89 100644
--- a/crates/daphne/src/testing/mod.rs
+++ b/crates/daphne/src/testing/mod.rs
@@ -340,13 +340,13 @@ impl AggregationJobTest {
         let (leader_state, agg_job_init_req) = self.produce_agg_job_req(&agg_param, reports);
 
         let (leader_agg_span, helper_agg_span) = {
-            let (helper_agg_span, AggregationJobResp::Ready { transitions }) =
+            let (helper_agg_span, AggregationJobResp::Ready { prep_resps }) =
                 self.handle_agg_job_req(agg_job_init_req)
             else {
                 panic!("testing should not be async")
             };
             let leader_agg_span =
-                self.consume_agg_job_resp(leader_state, ReadyAggregationJobResp { transitions });
+                self.consume_agg_job_resp(leader_state, ReadyAggregationJobResp { prep_resps });
             (leader_agg_span, helper_agg_span)
         };