feat(task): record reporter uuid when commiting task

Author: BobAnkh

netmito/src/api/workers.rs

@@ -150,7 +150,7 @@ pub async fn worker_report_task(
     State(pool): State<InfraPool>,
     Json(ReportTaskReq { id, op }): Json<ReportTaskReq>,
 ) -> ApiResult<Json<ReportTaskResp>> {
-    match service::worker::report_task(w.id, id, op, &pool).await {
+    match service::worker::report_task(w.id, w.uuid, id, op, &pool).await {
         Ok(url) => Ok(Json(ReportTaskResp { url })),
         Err(Error::ApiError(e)) => Err(e),
         Err(e) => {

netmito/src/client/interactive.rs

@@ -108,6 +108,9 @@ pub(crate) fn output_parsed_task_info(info: &ParsedTaskQueryInfo) {
     if let Some(downstream_task_uuid) = info.downstream_task_uuid {
         tracing::info!("Downstream Task UUID: {:?}", downstream_task_uuid);
     }
+    if let Some(reporter_uuid) = info.reporter_uuid {
+        tracing::info!("Reporter UUID: {}", reporter_uuid);
+    }
 }
 
 pub(crate) fn output_task_info(info: &TaskQueryInfo) {
@@ -140,6 +143,9 @@ pub(crate) fn output_task_info(info: &TaskQueryInfo) {
     if let Some(downstream_task_uuid) = info.downstream_task_uuid {
         tracing::info!("Downstream Task UUID: {:?}", downstream_task_uuid);
     }
+    if let Some(reporter_uuid) = info.reporter_uuid {
+        tracing::info!("Reporter UUID: {}", reporter_uuid);
+    }
 }
 
 pub(crate) fn output_worker_list_info<T: std::fmt::Display>(

netmito/src/config/client/artifacts.rs

@@ -103,6 +103,9 @@ pub struct DownloadArtifactsByFilterArgs {
     /// The priority of the tasks, support operators like `=`(default), `!=`, `<`, `<=`, `>`, `>=`
     #[arg(short, long)]
     pub priority: Option<String>,
+    /// Filter by reporter worker UUID (only returns artifacts from completed tasks reported by this worker)
+    #[arg(long)]
+    pub reporter_uuid: Option<Uuid>,
     /// Specify the directory to download artifacts
     #[arg(short, long = "output")]
     pub output_dir: Option<PathBuf>,
@@ -142,6 +145,7 @@ pub struct DownloadArtifactsByListArgs {
 impl From<DownloadArtifactsByFilterArgs> for ArtifactsDownloadByFilterReq {
     fn from(args: DownloadArtifactsByFilterArgs) -> Self {
         Self {
+            reporter_uuid: args.reporter_uuid,
             creator_usernames: if args.creators.is_empty() {
                 None
             } else {
@@ -205,6 +209,9 @@ pub struct DeleteArtifactsByFilterArgs {
     /// The priority of the tasks, support operators like `=`(default), `!=`, `<`, `<=`, `>`, `>=`
     #[arg(short, long)]
     pub priority: Option<String>,
+    /// Filter by reporter worker UUID (only deletes artifacts from completed tasks reported by this worker)
+    #[arg(long)]
+    pub reporter_uuid: Option<Uuid>,
 }
 
 #[derive(Serialize, Debug, Deserialize, Args, Clone)]
@@ -220,6 +227,7 @@ pub struct DeleteArtifactsByListArgs {
 impl From<DeleteArtifactsByFilterArgs> for ArtifactsDeleteByFilterReq {
     fn from(args: DeleteArtifactsByFilterArgs) -> Self {
         Self {
+            reporter_uuid: args.reporter_uuid,
             creator_usernames: if args.creators.is_empty() {
                 None
             } else {

netmito/src/config/client/tasks.rs

@@ -106,6 +106,9 @@ pub struct QueryTasksArgs {
     /// The priority of the tasks, support operators like `=`(default), `!=`, `<`, `<=`, `>`, `>=`
     #[arg(short, long)]
     pub priority: Option<String>,
+    /// Filter by reporter worker UUID (only returns completed tasks reported by this worker)
+    #[arg(long)]
+    pub reporter_uuid: Option<Uuid>,
     /// The limit of the tasks to query
     #[arg(long)]
     pub limit: Option<u64>,
@@ -200,6 +203,7 @@ pub struct ChangeTaskArgs {
 impl From<QueryTasksArgs> for TasksQueryReq {
     fn from(args: QueryTasksArgs) -> Self {
         Self {
+            reporter_uuid: args.reporter_uuid,
             creator_usernames: if args.creators.is_empty() {
                 None
             } else {

netmito/src/entity/active_tasks.rs

@@ -62,6 +62,8 @@ impl Related<super::users::Entity> for Entity {
 
 impl ActiveModelBehavior for ActiveModel {}
 
+/// This conversion is used when archiving an active task.
+/// But do remember to set reporter_uuid separately after conversion
 impl From<Model> for super::archived_tasks::Model {
     fn from(task: Model) -> super::archived_tasks::Model {
         Self {
@@ -82,6 +84,7 @@ impl From<Model> for super::archived_tasks::Model {
             result: task.result,
             upstream_task_uuid: task.upstream_task_uuid,
             downstream_task_uuid: task.downstream_task_uuid,
+            reporter_uuid: None,
         }
     }
 }

netmito/src/entity/archived_tasks.rs

@@ -26,6 +26,7 @@ pub struct Model {
     pub result: Option<Json>,
     pub upstream_task_uuid: Option<Uuid>,
     pub downstream_task_uuid: Option<Uuid>,
+    pub reporter_uuid: Option<Uuid>,
 }
 
 #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]

netmito/src/migration/m20251031_000001_add_reporter_uuid_to_archived_tasks.rs

@@ -0,0 +1,55 @@
+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(ArchivedTasks::Table)
+                    .add_column_if_not_exists(
+                        ColumnDef::new(ArchivedTasks::ReporterUuid).uuid().null(),
+                    )
+                    .to_owned(),
+            )
+            .await?;
+        manager
+            .create_index(
+                sea_query::Index::create()
+                    .if_not_exists()
+                    .name("idx_archived_tasks-reporter_uuid")
+                    .table(ArchivedTasks::Table)
+                    .col(ArchivedTasks::ReporterUuid)
+                    .and_where(Expr::col(ArchivedTasks::ReporterUuid).is_not_null())
+                    .to_owned(),
+            )
+            .await
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .drop_index(
+                sea_query::Index::drop()
+                    .name("idx_archived_tasks-reporter_uuid")
+                    .to_owned(),
+            )
+            .await?;
+        manager
+            .alter_table(
+                Table::alter()
+                    .table(ArchivedTasks::Table)
+                    .drop_column(ArchivedTasks::ReporterUuid)
+                    .to_owned(),
+            )
+            .await
+    }
+}
+
+#[derive(DeriveIden)]
+enum ArchivedTasks {
+    Table,
+    ReporterUuid,
+}

netmito/src/migration/mod.rs

@@ -6,6 +6,7 @@ mod m20250910_192001_add_performance_indices;
 mod m20250910_205347_add_labels_to_workers;
 mod m20250911_025409_add_gin_indices_on_tags;
 mod m20250915_111689_add_tasks_trigger;
+mod m20251031_000001_add_reporter_uuid_to_archived_tasks;
 
 pub struct Migrator;
 
@@ -19,6 +20,7 @@ impl MigratorTrait for Migrator {
             Box::new(m20250910_205347_add_labels_to_workers::Migration),
             Box::new(m20250911_025409_add_gin_indices_on_tags::Migration),
             Box::new(m20250915_111689_add_tasks_trigger::Migration),
+            Box::new(m20251031_000001_add_reporter_uuid_to_archived_tasks::Migration),
         ]
     }
 }

netmito/src/schema.rs

@@ -258,6 +258,7 @@ pub struct TaskQueryInfo {
     pub result: Option<serde_json::Value>,
     pub upstream_task_uuid: Option<Uuid>,
     pub downstream_task_uuid: Option<Uuid>,
+    pub reporter_uuid: Option<Uuid>,
 }
 
 #[derive(Debug, Serialize, Deserialize, Clone)]
@@ -277,6 +278,7 @@ pub struct ParsedTaskQueryInfo {
     pub result: Option<TaskResultSpec>,
     pub upstream_task_uuid: Option<Uuid>,
     pub downstream_task_uuid: Option<Uuid>,
+    pub reporter_uuid: Option<Uuid>,
 }
 
 /// Each field in the query request is optional, and the server will return all tasks if no field is specified.
@@ -291,6 +293,8 @@ pub struct TasksQueryReq {
     pub states: Option<HashSet<TaskState>>,
     pub exit_status: Option<String>,
     pub priority: Option<String>,
+    /// Set reporter_uuid will automatically exclude all non-completed tasks.
+    pub reporter_uuid: Option<Uuid>,
     pub limit: Option<u64>,
     pub offset: Option<u64>,
     pub count: bool,
@@ -556,8 +560,9 @@ pub struct WorkerShutdown {
     pub op: Option<WorkerShutdownOp>,
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone)]
+#[derive(Debug, Serialize, Deserialize, Clone, Default)]
 pub enum WorkerShutdownOp {
+    #[default]
     #[serde(alias = "graceful")]
     Graceful,
     #[serde(alias = "force")]
@@ -575,6 +580,8 @@ pub struct ArtifactsDownloadByFilterReq {
     pub states: Option<HashSet<TaskState>>,
     pub exit_status: Option<String>,
     pub priority: Option<String>,
+    /// Set reporter_uuid will automatically exclude all non-completed tasks.
+    pub reporter_uuid: Option<Uuid>,
     pub content_type: ArtifactContentType,
 }
 
@@ -638,6 +645,8 @@ pub struct ArtifactsDeleteByFilterReq {
     pub states: Option<HashSet<TaskState>>,
     pub exit_status: Option<String>,
     pub priority: Option<String>,
+    /// Set reporter_uuid will automatically exclude all non-completed tasks.
+    pub reporter_uuid: Option<Uuid>,
     pub content_type: ArtifactContentType,
 }
 
@@ -702,12 +711,6 @@ pub struct TasksSubmitResp {
     pub results: Vec<Result<SubmitTaskResp, crate::error::ErrorMsg>>,
 }
 
-impl Default for WorkerShutdownOp {
-    fn default() -> Self {
-        Self::Graceful
-    }
-}
-
 impl TaskSpec {
     pub fn new<T, I, P, Q, V, U>(
         args: I,

netmito/src/service/auth/mod.rs

@@ -42,6 +42,7 @@ pub struct AuthAdminUser {
 #[derive(Debug, Clone)]
 pub struct AuthWorker {
     pub id: i64,
+    pub uuid: Uuid,
 }
 
 pub(crate) fn get_and_prompt_username(
@@ -344,5 +345,8 @@ async fn worker_auth(db: &DatabaseConnection, bearer: &Bearer) -> Result<AuthWor
         .await
         .map_err(|_| AuthError::WrongCredentials)?
         .ok_or(AuthError::WrongCredentials)?;
-    Ok(AuthWorker { id: worker.id })
+    Ok(AuthWorker {
+        id: worker.id,
+        uuid,
+    })
 }