Aggregator API: return 200 with zero counters for fresh task (#2753)

aggregator_api/src/tests.rs

@@ -872,21 +872,36 @@ async fn get_task_upload_metrics() {
                 let task_id = *task.id();
                 tx.put_aggregator_task(&task).await.unwrap();
 
-                tx.increment_task_upload_counter(
-                    &task_id,
-                    1,
-                    &TaskUploadCounter::new_with_values(0, 0, 2, 4, 6, 100, 25, 12),
-                )
-                .await
-                .unwrap();
-
                 Ok(task_id)
             })
         })
         .await
         .unwrap();
 
+    // Verify: requesting metrics on a fresh task returns zeroes.
+    assert_response!(
+        get(&format!("/tasks/{}/metrics/uploads", &task_id))
+            .with_request_header("Authorization", format!("Bearer {AUTH_TOKEN}"))
+            .with_request_header("Accept", CONTENT_TYPE)
+            .run_async(&handler)
+            .await,
+        Status::Ok,
+        serde_json::to_string(&GetTaskUploadMetricsResp(TaskUploadCounter::default())).unwrap(),
+    );
+
     // Verify: requesting metrics on a task returns the correct result.
+    ds.run_unnamed_tx(|tx| {
+        Box::pin(async move {
+            tx.increment_task_upload_counter(
+                &task_id,
+                1,
+                &TaskUploadCounter::new_with_values(0, 0, 2, 4, 6, 100, 25, 12),
+            )
+            .await
+        })
+    })
+    .await
+    .unwrap();
     assert_response!(
         get(&format!("/tasks/{}/metrics/uploads", &task_id))
             .with_request_header("Authorization", format!("Bearer {AUTH_TOKEN}"))

aggregator_core/src/datastore.rs

@@ -4976,36 +4976,37 @@ impl<C: Clock> Transaction<'_, C> {
         let stmt = self
             .prepare_cached(
                 "SELECT
-                    SUM(interval_collected)::BIGINT AS interval_collected,
-                    SUM(report_decode_failure)::BIGINT AS report_decode_failure,
-                    SUM(report_decrypt_failure)::BIGINT AS report_decrypt_failure,
-                    SUM(report_expired)::BIGINT AS report_expired,
-                    SUM(report_outdated_key)::BIGINT AS report_outdated_key,
-                    SUM(report_success)::BIGINT AS report_success,
-                    SUM(report_too_early)::BIGINT AS report_too_early,
-                    SUM(task_expired)::BIGINT AS task_expired
+                    tasks.id,
+                    COALESCE(SUM(interval_collected)::BIGINT, 0) AS interval_collected,
+                    COALESCE(SUM(report_decode_failure)::BIGINT, 0) AS report_decode_failure,
+                    COALESCE(SUM(report_decrypt_failure)::BIGINT, 0) AS report_decrypt_failure,
+                    COALESCE(SUM(report_expired)::BIGINT, 0) AS report_expired,
+                    COALESCE(SUM(report_outdated_key)::BIGINT, 0) AS report_outdated_key,
+                    COALESCE(SUM(report_success)::BIGINT, 0) AS report_success,
+                    COALESCE(SUM(report_too_early)::BIGINT, 0) AS report_too_early,
+                    COALESCE(SUM(task_expired)::BIGINT, 0) AS task_expired
                 FROM task_upload_counters
-                WHERE task_id = (SELECT id FROM tasks WHERE task_id = $1)",
+                RIGHT JOIN tasks on tasks.id = task_upload_counters.task_id
+                WHERE tasks.task_id = $1
+                GROUP BY tasks.id",
             )
             .await?;
 
-        let row = self.query_one(&stmt, &[task_id.as_ref()]).await?;
-        let interval_collected = row.get_nullable_bigint_and_convert("interval_collected")?;
-        Ok(match interval_collected {
-            Some(interval_collected) => Some(TaskUploadCounter {
-                interval_collected,
-                // The remaining columns should exist if the first one did, due to a DEFAULT 0
-                // clause, so we don't need to treat these as nullable.
-                report_decode_failure: row.get_bigint_and_convert("report_decode_failure")?,
-                report_decrypt_failure: row.get_bigint_and_convert("report_decrypt_failure")?,
-                report_expired: row.get_bigint_and_convert("report_expired")?,
-                report_outdated_key: row.get_bigint_and_convert("report_outdated_key")?,
-                report_success: row.get_bigint_and_convert("report_success")?,
-                report_too_early: row.get_bigint_and_convert("report_too_early")?,
-                task_expired: row.get_bigint_and_convert("task_expired")?,
-            }),
-            None => None,
-        })
+        self.query_opt(&stmt, &[task_id.as_ref()])
+            .await?
+            .map(|row| {
+                Ok(TaskUploadCounter {
+                    interval_collected: row.get_bigint_and_convert("interval_collected")?,
+                    report_decode_failure: row.get_bigint_and_convert("report_decode_failure")?,
+                    report_decrypt_failure: row.get_bigint_and_convert("report_decrypt_failure")?,
+                    report_expired: row.get_bigint_and_convert("report_expired")?,
+                    report_outdated_key: row.get_bigint_and_convert("report_outdated_key")?,
+                    report_success: row.get_bigint_and_convert("report_success")?,
+                    report_too_early: row.get_bigint_and_convert("report_too_early")?,
+                    task_expired: row.get_bigint_and_convert("task_expired")?,
+                })
+            })
+            .transpose()
     }
 
     /// Add a `TaskUploadCounter` to the counter associated with the given [`TaskId`]. This is sharded,

aggregator_core/src/datastore/tests.rs

@@ -7611,9 +7611,14 @@ async fn roundtrip_task_upload_counter(ephemeral_datastore: EphemeralDatastore)
         .run_unnamed_tx(|tx| {
             let task_id = *task.id();
             Box::pin(async move {
-                let counter = tx.get_task_upload_counter(&task_id).await.unwrap();
+                // Returns None for non-existent task.
+                let counter = tx.get_task_upload_counter(&random()).await.unwrap();
                 assert_eq!(counter, None);
 
+                // Returns Some for a task that has just been created and has no counters.
+                let counter = tx.get_task_upload_counter(&task_id).await.unwrap();
+                assert_eq!(counter, Some(TaskUploadCounter::default()));
+
                 let ord = thread_rng().gen_range(0..32);
                 tx.increment_task_upload_counter(
                     &task_id,