EventPayload::TaskCanceled renamed to TasksCanceled

Author: spirali

crates/hyperqueue/src/client/commands/journal/output.rs

@@ -87,7 +87,7 @@ fn format_payload(event: EventPayload) -> serde_json::Value {
             "job": task_id.job_id(),
             "task": task_id.job_task_id(),
         }),
-        EventPayload::TaskCanceled { task_ids } => json!({
+        EventPayload::TasksCanceled { task_ids } => json!({
             "type": "task-canceled",
             "tasks": task_ids,
         }),

crates/hyperqueue/src/server/client/mod.rs

@@ -354,7 +354,7 @@ fn reconstruct_historical_events(
                 }
                 JobTaskState::Canceled { cancelled_date, .. } => {
                     if let Some(task_ids) = events.last_mut().and_then(|e| {
-                        if let EventPayload::TaskCanceled { task_ids, .. } = &mut e.payload {
+                        if let EventPayload::TasksCanceled { task_ids, .. } = &mut e.payload {
                             (e.time == *cancelled_date).then_some(task_ids)
                         } else {
                             None
@@ -364,7 +364,7 @@ fn reconstruct_historical_events(
                     } else {
                         events.push(Event::at(
                             *cancelled_date,
-                            EventPayload::TaskCanceled {
+                            EventPayload::TasksCanceled {
                                 task_ids: vec![TaskId::new(job.job_id, *id)],
                             },
                         ));

crates/hyperqueue/src/server/event/journal/prune.rs

@@ -28,7 +28,7 @@ pub(crate) fn prune_journal(
             | EventPayload::TaskFailed { task_id, .. } => {
                 live_job_ids.contains(&task_id.job_id()).then_some(event)
             }
-            EventPayload::TaskCanceled { task_ids, .. } => {
+            EventPayload::TasksCanceled { task_ids, .. } => {
                 task_ids.retain(|id| live_job_ids.contains(&id.job_id()));
                 (!task_ids.is_empty()).then_some(event)
             }

crates/hyperqueue/src/server/event/payload.rs

@@ -46,8 +46,8 @@ pub enum EventPayload {
         task_id: TaskId,
         error: String,
     },
-    /// Task has been canceled
-    TaskCanceled {
+    /// Tasks has been canceled; for performance and correctness reason, this even is batched.
+    TasksCanceled {
         task_ids: Vec<TaskId>,
     },
     /// New allocation queue has been created

crates/hyperqueue/src/server/event/streamer.rs

@@ -138,7 +138,7 @@ impl EventStreamer {
 
     pub fn on_task_canceled(&self, task_ids: Vec<TaskId>, now: DateTime<Utc>) {
         self.send_event(
-            EventPayload::TaskCanceled { task_ids },
+            EventPayload::TasksCanceled { task_ids },
             Some(now),
             ForwardMode::StreamAndPersist,
         );

crates/hyperqueue/src/server/restore.rs

@@ -290,7 +290,7 @@ impl StateRestorer {
                         }
                     }
                 }
-                EventPayload::TaskCanceled { task_ids } => {
+                EventPayload::TasksCanceled { task_ids } => {
                     log::debug!("Replaying: TaskCanceled {task_ids:?}");
                     for task_id in task_ids {
                         if let Some(job) = self.jobs.get_mut(&task_id.job_id()) {