Task expiration and deletion

cli/src/tasks.rs

@@ -1,7 +1,10 @@
+use std::time::SystemTime;
+
 use crate::{CliResult, DetermineAccountId, Error, Output};
 use clap::Subcommand;
 use divviup_client::{DivviupClient, Histogram, NewTask, Uuid, Vdaf};
-use humantime::Duration;
+use humantime::{Duration, Timestamp};
+use time::OffsetDateTime;
 
 #[derive(clap::ValueEnum, Clone, Debug)]
 pub enum VdafName {
@@ -52,6 +55,18 @@ pub enum TaskAction {
     /// rename a task
     Rename { task_id: String, name: String },
 
+    /// delete a task
+    Delete { task_id: String },
+
+    /// set the expiration date of a task
+    SetExpiration {
+        task_id: String,
+        /// the date and time to set to.
+        ///
+        /// if omitted, unset the expiration. the format is RFC 3339.
+        expiration: Option<Timestamp>,
+    },
+
     /// retrieve the collector auth tokens for a task
     CollectorAuthTokens { task_id: String },
 }
@@ -149,6 +164,20 @@ impl TaskAction {
             TaskAction::CollectorAuthTokens { task_id } => {
                 output.display(client.task_collector_auth_tokens(&task_id).await?)
             }
+            TaskAction::Delete { task_id } => client.delete_task(&task_id).await?,
+            TaskAction::SetExpiration {
+                task_id,
+                expiration,
+            } => output.display(
+                client
+                    .set_task_expiration(
+                        &task_id,
+                        expiration
+                            .map(|e| OffsetDateTime::from(Into::<SystemTime>::into(e)))
+                            .as_ref(),
+                    )
+                    .await?,
+            ),
         }
 
         Ok(())

client/src/lib.rs

@@ -24,6 +24,7 @@ use base64::{engine::general_purpose::STANDARD, Engine};
 use serde::{de::DeserializeOwned, Serialize};
 use serde_json::json;
 use std::{fmt::Display, future::Future, pin::Pin};
+use time::format_description::well_known::Rfc3339;
 use trillium_http::{HeaderName, HeaderValues};
 
 pub use account::Account;
@@ -269,6 +270,24 @@ impl DivviupClient {
             .await
     }
 
+    pub async fn set_task_expiration(
+        &self,
+        task_id: &str,
+        expiration: Option<&OffsetDateTime>,
+    ) -> ClientResult<Task> {
+        self.patch(
+            &format!("api/tasks/{task_id}"),
+            &json!({
+                "expiration": expiration.map(|e| e.format(&Rfc3339)).transpose()?
+            }),
+        )
+        .await
+    }
+
+    pub async fn delete_task(&self, task_id: &str) -> ClientResult<()> {
+        self.delete(&format!("api/tasks/{task_id}")).await
+    }
+
     pub async fn api_tokens(&self, account_id: Uuid) -> ClientResult<Vec<ApiToken>> {
         self.get(&format!("api/accounts/{account_id}/api_tokens"))
             .await
@@ -379,6 +398,9 @@ pub enum Error {
 
     #[error(transparent)]
     Codec(#[from] CodecError),
+
+    #[error("time formatting error: {0}")]
+    TimeFormat(#[from] time::error::Format),
 }
 
 pub trait ClientConnExt: Sized {

client/src/task.rs

@@ -15,6 +15,8 @@ pub struct Task {
     pub created_at: OffsetDateTime,
     #[serde(with = "time::serde::rfc3339")]
     pub updated_at: OffsetDateTime,
+    #[serde(with = "time::serde::rfc3339::option")]
+    pub deleted_at: Option<OffsetDateTime>,
     pub time_precision_seconds: u32,
     #[deprecated = "Not populated. Will be removed in a future release."]
     pub report_count: u32,

client/tests/integration/tasks.rs

@@ -86,8 +86,49 @@ async fn rename_task(app: Arc<DivviupApi>, account: Account, client: DivviupClie
     let task = fixtures::task(&app, &account).await;
     let name = fixtures::random_name();
     let response = client.rename_task(&task.id, &name).await?;
-    assert_eq!(&response.name, &name);
-    assert_eq!(task.reload(app.db()).await?.unwrap().name, name);
+    assert_eq!(response.name, name);
+    assert_eq!(response.expiration, task.expiration);
+    let task_reload = task.reload(app.db()).await?.unwrap();
+    assert_eq!(task_reload.name, name);
+    assert_eq!(task_reload.expiration, task.expiration);
+    Ok(())
+}
+
+#[test(harness = with_configured_client)]
+async fn set_task_expiration(
+    app: Arc<DivviupApi>,
+    account: Account,
+    client: DivviupClient,
+) -> TestResult {
+    let task = fixtures::task(&app, &account).await;
+    let now = OffsetDateTime::now_utc();
+    let response = client.set_task_expiration(&task.id, Some(&now)).await?;
+    assert_eq!(response.name, task.name);
+    assert_eq!(response.expiration, Some(now));
+    let task_reload = task.reload(app.db()).await?.unwrap();
+    assert_eq!(task_reload.name, task.name);
+    assert_eq!(task_reload.expiration, Some(now));
+
+    let response = client.set_task_expiration(&task.id, None).await?;
+    assert_eq!(response.name, task.name);
+    assert_eq!(response.expiration, None);
+    let task_reload = task.reload(app.db()).await?.unwrap();
+    assert_eq!(task_reload.name, task.name);
+    assert_eq!(task_reload.expiration, None);
+    Ok(())
+}
+
+#[test(harness = with_configured_client)]
+async fn delete_task(app: Arc<DivviupApi>, account: Account, client: DivviupClient) -> TestResult {
+    let task = fixtures::task(&app, &account).await;
+
+    let response_tasks = client.tasks(account.id).await?;
+    assert!(!response_tasks.is_empty());
+
+    client.delete_task(&task.id).await?;
+
+    let response_tasks = client.tasks(account.id).await?;
+    assert!(response_tasks.is_empty());
     Ok(())
 }
 

migration/src/lib.rs

@@ -24,6 +24,7 @@ mod m20231012_225001_rename_hpke_configs_to_collector_credentials;
 mod m20231012_233117_add_token_hash_to_collector_credential;
 mod m20240214_215101_upload_metrics;
 mod m20240411_195358_time_bucketed_fixed_size;
+mod m20240416_172920_task_deleted_at;
 
 pub struct Migrator;
 
@@ -55,6 +56,7 @@ impl MigratorTrait for Migrator {
             Box::new(m20231012_233117_add_token_hash_to_collector_credential::Migration),
             Box::new(m20240214_215101_upload_metrics::Migration),
             Box::new(m20240411_195358_time_bucketed_fixed_size::Migration),
+            Box::new(m20240416_172920_task_deleted_at::Migration),
         ]
     }
 }

migration/src/m20240416_172920_task_deleted_at.rs

@@ -0,0 +1,39 @@
+use sea_orm_migration::prelude::*;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .alter_table(
+                Table::alter()
+                    .table(Task::Table)
+                    .add_column(
+                        ColumnDef::new(Task::DeletedAt)
+                            .timestamp_with_time_zone()
+                            .null(),
+                    )
+                    .to_owned(),
+            )
+            .await
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .alter_table(
+                TableAlterStatement::new()
+                    .table(Task::Table)
+                    .drop_column(Task::DeletedAt)
+                    .to_owned(),
+            )
+            .await
+    }
+}
+
+#[derive(DeriveIden)]
+enum Task {
+    Table,
+    DeletedAt,
+}

src/api_mocks/aggregator_api.rs

@@ -3,7 +3,7 @@ use crate::{
     clients::aggregator_client::api_types::{
         AggregatorApiConfig, AggregatorVdaf, AuthenticationToken, HpkeAeadId, HpkeConfig,
         HpkeKdfId, HpkeKemId, HpkePublicKey, JanusDuration, QueryType, Role, TaskCreate, TaskId,
-        TaskIds, TaskResponse, TaskUploadMetrics,
+        TaskIds, TaskPatch, TaskResponse, TaskUploadMetrics,
     },
     entity::aggregator::{Feature, Features},
 };
@@ -41,6 +41,7 @@ pub fn mock() -> impl Handler {
             )
             .post("/tasks", api(post_task))
             .get("/tasks/:task_id", api(get_task))
+            .patch("/tasks/:task_id", api(patch_task))
             .get("/task_ids", api(task_ids))
             .delete("/tasks/:task_id", Status::Ok)
             .get(
@@ -109,6 +110,28 @@ async fn post_task(
     (State(task_create.clone()), Json(task_response(task_create)))
 }
 
+async fn patch_task(conn: &mut Conn, Json(patch): Json<TaskPatch>) -> Json<TaskResponse> {
+    let task_id = conn.param("task_id").unwrap();
+    Json(TaskResponse {
+        task_id: task_id.parse().unwrap(),
+        peer_aggregator_endpoint: "https://_".parse().unwrap(),
+        query_type: QueryType::TimeInterval,
+        vdaf: AggregatorVdaf::Prio3Count,
+        role: Role::Leader,
+        vdaf_verify_key: random_chars(10),
+        max_batch_query_count: 100,
+        task_expiration: patch.task_expiration,
+        report_expiry_age: None,
+        min_batch_size: 1000,
+        time_precision: JanusDuration::from_seconds(60),
+        tolerable_clock_skew: JanusDuration::from_seconds(60),
+        collector_hpke_config: random_hpke_config(),
+        aggregator_auth_token: Some(AuthenticationToken::new(random_chars(32))),
+        collector_auth_token: Some(AuthenticationToken::new(random_chars(32))),
+        aggregator_hpke_configs: repeat_with(random_hpke_config).take(5).collect(),
+    })
+}
+
 pub fn task_response(task_create: TaskCreate) -> TaskResponse {
     let task_id = TaskId::try_from(
         Sha256::digest(URL_SAFE_NO_PAD.decode(task_create.vdaf_verify_key).unwrap()).as_slice(),

src/clients/aggregator_client.rs

@@ -3,11 +3,14 @@ use crate::{
     entity::{task::ProvisionableTask, Aggregator},
     handler::Error,
 };
+use janus_messages::Time as JanusTime;
 use serde::{de::DeserializeOwned, Serialize};
 use trillium_client::{Client, KnownHeaderName};
 use url::Url;
 pub mod api_types;
-pub use api_types::{AggregatorApiConfig, TaskCreate, TaskIds, TaskResponse, TaskUploadMetrics};
+pub use api_types::{
+    AggregatorApiConfig, TaskCreate, TaskIds, TaskPatch, TaskResponse, TaskUploadMetrics,
+};
 
 const CONTENT_TYPE: &str = "application/vnd.janus.aggregator+json;version=0.1";
 
@@ -83,15 +86,26 @@ impl AggregatorClient {
         self.get(&format!("tasks/{task_id}/metrics/uploads")).await
     }
 
-    pub async fn delete_task(&self, task_id: &str) -> Result<(), ClientError> {
-        self.delete(&format!("tasks/{task_id}")).await
-    }
-
     pub async fn create_task(&self, task: &ProvisionableTask) -> Result<TaskResponse, Error> {
         let task_create = TaskCreate::build(&self.aggregator, task)?;
         self.post("tasks", &task_create).await.map_err(Into::into)
     }
 
+    pub async fn update_task_expiration(
+        &self,
+        task_id: &str,
+        expiration: Option<JanusTime>,
+    ) -> Result<TaskResponse, Error> {
+        self.patch(
+            &format!("tasks/{task_id}"),
+            &TaskPatch {
+                task_expiration: expiration,
+            },
+        )
+        .await
+        .map_err(Into::into)
+    }
+
     // private below here
 
     async fn get<T: DeserializeOwned>(&self, path: &str) -> Result<T, ClientError> {
@@ -120,8 +134,19 @@ impl AggregatorClient {
             .map_err(ClientError::from)
     }
 
-    async fn delete(&self, path: &str) -> Result<(), ClientError> {
-        let _ = self.client.delete(path).success_or_client_error().await?;
-        Ok(())
+    async fn patch<T: DeserializeOwned>(
+        &self,
+        path: &str,
+        body: &impl Serialize,
+    ) -> Result<T, ClientError> {
+        self.client
+            .patch(path)
+            .with_json_body(body)?
+            .with_request_header(KnownHeaderName::ContentType, CONTENT_TYPE)
+            .success_or_client_error()
+            .await?
+            .response_json()
+            .await
+            .map_err(ClientError::from)
     }
 }

src/clients/aggregator_client/api_types.rs

@@ -289,6 +289,11 @@ impl TaskCreate {
     }
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone, Copy)]
+pub struct TaskPatch {
+    pub task_expiration: Option<JanusTime>,
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct TaskResponse {
     pub task_id: TaskId,