Poll collect endpoint with GET.

crates/daphne-server/src/router/leader.rs

@@ -5,10 +5,11 @@ use std::sync::Arc;
 
 use axum::{
     body::HttpBody,
-    extract::State,
-    http::StatusCode,
+    extract::{Path, State},
+    http::{Request, StatusCode},
+    middleware::{from_fn, Next},
     response::{IntoResponse, Response},
-    routing::{post, put},
+    routing::{get, post, put},
 };
 use daphne::{
     constants::DapMediaType,
@@ -23,6 +24,28 @@ use super::{
     extractor::dap_sender::FROM_COLLECTOR, AxumDapResponse, DapRequestExtractor, DaphneService,
     UnauthenticatedDapRequestExtractor,
 };
+use futures::{future::BoxFuture, FutureExt};
+use serde::Deserialize;
+
+#[derive(Deserialize)]
+struct PathVersion {
+    #[serde(rename = "version")]
+    presented_version: DapVersion,
+}
+
+fn require_version<B: Send + 'static>(
+    expected_version: DapVersion,
+) -> impl Copy + Fn(Path<PathVersion>, Request<B>, Next<B>) -> BoxFuture<'static, Response> {
+    move |Path(PathVersion { presented_version }), req, next| {
+        async move {
+            if presented_version != expected_version {
+                return StatusCode::METHOD_NOT_ALLOWED.into_response();
+            }
+            next.run(req).await
+        }
+        .boxed()
+    }
+}
 
 pub(super) fn add_leader_routes<A, B>(router: super::Router<A, B>) -> super::Router<A, B>
 where
@@ -32,11 +55,25 @@ where
     B::Error: Send + Sync,
 {
     router
-        .route("/:version/tasks/:task_id/reports", put(upload_draft09))
-        .route("/:version/tasks/:task_id/reports", post(upload_draft13))
+        .route(
+            "/:version/tasks/:task_id/reports",
+            put(upload).layer(from_fn(require_version(DapVersion::Draft09))),
+        )
+        .route(
+            "/:version/tasks/:task_id/reports",
+            post(upload).layer(from_fn(require_version(DapVersion::Latest))),
+        )
         .route(
             "/:version/tasks/:task_id/collection_jobs/:collect_job_id",
-            put(start_collection_job).post(collect),
+            put(start_collection_job),
+        )
+        .route(
+            "/:version/tasks/:task_id/collection_jobs/:collect_job_id",
+            post(poll_collect).layer(from_fn(require_version(DapVersion::Draft09))),
+        )
+        .route(
+            "/:version/tasks/:task_id/collection_jobs/:collect_job_id",
+            get(poll_collect).layer(from_fn(require_version(DapVersion::Latest))),
         )
 }
 
@@ -47,7 +84,7 @@ where
         version = ?req.version,
     )
 )]
-async fn upload_draft13<A>(
+async fn upload<A>(
     State(app): State<Arc<A>>,
     UnauthenticatedDapRequestExtractor(req): UnauthenticatedDapRequestExtractor<
         messages::Report,
@@ -57,41 +94,12 @@ async fn upload_draft13<A>(
 where
     A: DapLeader + DaphneService + Send + Sync,
 {
-    if req.version == DapVersion::Draft09 {
-        return (
-            StatusCode::METHOD_NOT_ALLOWED,
-            format!("route not implemented for version {}", req.version),
-        )
-            .into_response();
-    }
     match leader::handle_upload_req(&*app, req).await {
         Ok(()) => StatusCode::OK.into_response(),
         Err(e) => AxumDapResponse::new_error(e, app.server_metrics()).into_response(),
     }
 }
 
-async fn upload_draft09<A>(
-    State(app): State<Arc<A>>,
-    UnauthenticatedDapRequestExtractor(req): UnauthenticatedDapRequestExtractor<
-        messages::Report,
-        resource::None,
-    >,
-) -> Response
-where
-    A: DapLeader + DaphneService + Send + Sync,
-{
-    if req.version != DapVersion::Draft09 {
-        return (
-            StatusCode::METHOD_NOT_ALLOWED,
-            format!("route not implemented for version {}", req.version),
-        )
-            .into_response();
-    }
-    match leader::handle_upload_req(&*app, req).await {
-        Ok(()) => StatusCode::OK.into_response(),
-        Err(e) => AxumDapResponse::new_error(e, app.server_metrics()).into_response(),
-    }
-}
 #[tracing::instrument(
     skip_all,
     fields(
@@ -123,7 +131,7 @@ where
         version = ?req.version,
     )
 )]
-async fn collect<A>(
+async fn poll_collect<A>(
     State(app): State<Arc<A>>,
     DapRequestExtractor(req): DapRequestExtractor<FROM_COLLECTOR, (), resource::CollectionJobId>,
 ) -> Response

crates/daphne-server/tests/e2e/e2e.rs

@@ -840,6 +840,86 @@ async fn leader_collect_abort_unknown_request(version: DapVersion) {
 
 async_test_versions! { leader_collect_abort_unknown_request }
 
+async fn leader_collect_back_compat(version: DapVersion) {
+    let t = TestRunner::default_with_version(version).await;
+    let batch_interval = t.batch_interval();
+
+    let client = t.http_client();
+    let hpke_config_list = t.get_hpke_configs(version, client).await.unwrap();
+    let path = t.upload_path();
+    let method = match version {
+        DapVersion::Draft09 => &Method::PUT,
+        DapVersion::Latest => &Method::POST,
+    };
+    let expected_status = 405;
+
+    // The reports are uploaded in the background.
+    let mut rng = thread_rng();
+    let mut time_min = u64::MAX;
+    let mut time_max = 0u64;
+    for _ in 0..t.task_config.min_batch_size {
+        let now = rng.gen_range(TestRunner::report_interval(&batch_interval));
+        time_min = min(time_min, now);
+        time_max = max(time_max, now);
+        t.leader_request_expect_ok(
+            client,
+            &path,
+            method,
+            DapMediaType::Report,
+            None,
+            t.task_config
+                .vdaf
+                .produce_report(
+                    &hpke_config_list,
+                    now,
+                    &t.task_id,
+                    DapMeasurement::U64(1),
+                    version,
+                )
+                .unwrap()
+                .get_encoded_with_param(&version)
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+    }
+
+    // Get the collect URI.
+    let agg_param = DapAggregationParam::Empty;
+    let collect_req = CollectionReq {
+        query: Query::TimeInterval { batch_interval },
+        agg_param: agg_param.get_encoded().unwrap(),
+    };
+    let collect_uri = t
+        .leader_post_collect(
+            client,
+            collect_req.get_encoded_with_param(&t.version).unwrap(),
+        )
+        .await
+        .unwrap();
+    println!("collect_uri: {collect_uri}");
+
+    let builder = match version {
+        DapVersion::Draft09 => client.get(collect_uri.as_str()),
+        DapVersion::Latest => client.post(collect_uri.as_str()),
+    };
+    let mut headers = reqwest::header::HeaderMap::new();
+    headers.insert(
+        reqwest::header::HeaderName::from_static(http_headers::DAP_AUTH_TOKEN),
+        reqwest::header::HeaderValue::from_str(&t.collector_bearer_token)
+            .expect("couldn't parse bearer token"),
+    );
+    let resp = builder
+        .headers(headers)
+        .send()
+        .await
+        .expect("failed to get a response");
+
+    assert_eq!(resp.status(), expected_status);
+}
+
+async_test_versions! { leader_collect_back_compat }
+
 async fn leader_collect_accept_global_config_max_batch_duration(version: DapVersion) {
     let t = TestRunner::default_with_version(version).await;
     let client = t.http_client();

crates/daphne-server/tests/e2e/test_runner.rs

@@ -669,7 +669,10 @@ impl TestRunner {
         url: &Url,
         token: &str,
     ) -> anyhow::Result<reqwest::Response> {
-        let builder = client.post(url.as_str());
+        let builder = match self.version {
+            DapVersion::Draft09 => client.post(url.as_str()),
+            DapVersion::Latest => client.get(url.as_str()),
+        };
         let mut headers = reqwest::header::HeaderMap::new();
         headers.insert(
             reqwest::header::HeaderName::from_static(http_headers::DAP_AUTH_TOKEN),