validate vdaf and query type are supported by both aggregators

src/clients/aggregator_client/api_types.rs

@@ -1,6 +1,6 @@
 use crate::{
     entity::{
-        aggregator::{QueryTypeNameSet, Role as AggregatorRole, VdafNameSet},
+        aggregator::{QueryTypeName, QueryTypeNameSet, Role as AggregatorRole, VdafNameSet},
         task::vdaf::{CountVec, Histogram, Sum, SumVec, Vdaf},
         Aggregator, ProvisionableTask,
     },
@@ -73,6 +73,15 @@ pub enum QueryType {
     FixedSize { max_batch_size: u64 },
 }
 
+impl QueryType {
+    pub fn name(&self) -> QueryTypeName {
+        match self {
+            QueryType::TimeInterval => QueryTypeName::TimeInterval,
+            QueryType::FixedSize { .. } => QueryTypeName::FixedSize,
+        }
+    }
+}
+
 impl From<QueryType> for Option<i64> {
     fn from(value: QueryType) -> Self {
         Option::<u64>::from(value).map(|u| u.try_into().unwrap())

src/entity/aggregator/query_type_name.rs

@@ -114,4 +114,8 @@ impl QueryTypeNameSet {
     pub fn intersect(&self, other: &QueryTypeNameSet) -> QueryTypeNameSet {
         self.0.intersection(&other.0).cloned().collect()
     }
+
+    pub fn contains(&self, name: &QueryTypeName) -> bool {
+        self.0.contains(name)
+    }
 }

src/entity/aggregator/vdaf_name.rs

@@ -125,4 +125,8 @@ impl VdafNameSet {
     pub fn intersect(&self, other: &VdafNameSet) -> VdafNameSet {
         self.0.intersection(&other.0).cloned().collect()
     }
+
+    pub fn contains(&self, name: &VdafName) -> bool {
+        self.0.contains(name)
+    }
 }

src/entity/task/new_task.rs

@@ -1,6 +1,6 @@
 use super::*;
 use crate::{
-    clients::aggregator_client::api_types::{Decode, HpkeConfig},
+    clients::aggregator_client::api_types::{Decode, HpkeConfig, QueryType},
     entity::{aggregator::Role, Account, Aggregator, Aggregators},
     handler::Error,
 };
@@ -11,7 +11,7 @@ use base64::{
 use rand::Rng;
 use sha2::{Digest, Sha256};
 use std::io::Cursor;
-use validator::ValidationErrors;
+use validator::{ValidationErrors, ValidationErrorsKind};
 
 fn in_the_future(time: &OffsetDateTime) -> Result<(), ValidationError> {
     if time < &OffsetDateTime::now_utc() {
@@ -184,6 +184,40 @@ impl NewTask {
         }
     }
 
+    fn validate_vdaf_is_supported(
+        &self,
+        leader: &Aggregator,
+        helper: &Aggregator,
+        errors: &mut ValidationErrors,
+    ) {
+        let Some(vdaf) = self.vdaf.as_ref() else {
+            return;
+        };
+        let name = vdaf.name();
+        if leader.vdafs.contains(&name) && helper.vdafs.contains(&name) {
+            return;
+        }
+        if let ValidationErrorsKind::Struct(errors) = errors
+            .errors_mut()
+            .entry("vdaf")
+            .or_insert_with(|| ValidationErrorsKind::Struct(Box::new(ValidationErrors::new())))
+        {
+            errors.add("type", ValidationError::new("not-supported"));
+        }
+    }
+
+    fn validate_query_type_is_supported(
+        &self,
+        leader: &Aggregator,
+        helper: &Aggregator,
+        errors: &mut ValidationErrors,
+    ) {
+        let name = QueryType::from(self.max_batch_size).name();
+        if !leader.query_types.contains(&name) || !helper.query_types.contains(&name) {
+            errors.add("max_batch_size", ValidationError::new("not-supported"));
+        }
+    }
+
     pub async fn validate(
         &self,
         account: Account,
@@ -193,6 +227,10 @@ impl NewTask {
         self.validate_min_lte_max(&mut errors);
         let hpke_config = self.validate_hpke_config(&mut errors);
         let aggregators = self.validate_aggregators(&account, db, &mut errors).await;
+        if let Some((leader, helper)) = aggregators.as_ref() {
+            self.validate_vdaf_is_supported(leader, helper, &mut errors);
+            self.validate_query_type_is_supported(leader, helper, &mut errors);
+        }
 
         if errors.is_empty() {
             // Unwrap safety: All of these unwraps below have previously

src/entity/task/vdaf.rs

@@ -1,4 +1,4 @@
-use crate::json_newtype;
+use crate::{entity::aggregator::VdafName, json_newtype};
 use serde::{Deserialize, Serialize};
 use validator::{Validate, ValidationError, ValidationErrors};
 
@@ -72,6 +72,19 @@ pub enum Vdaf {
     Unrecognized,
 }
 
+impl Vdaf {
+    pub fn name(&self) -> VdafName {
+        match self {
+            Vdaf::Count => VdafName::Prio3Count,
+            Vdaf::Histogram(_) => VdafName::Prio3Histogram,
+            Vdaf::Sum(_) => VdafName::Prio3Sum,
+            Vdaf::CountVec(_) => VdafName::Prio3Count,
+            Vdaf::SumVec(_) => VdafName::Prio3SumVec,
+            Vdaf::Unrecognized => VdafName::Other("unsupported".into()),
+        }
+    }
+}
+
 json_newtype!(Vdaf);
 
 impl Validate for Vdaf {