refactor(job): remove cancel_job API

Consumers handle cancellation at their own layer, so the crate-level
cancel_job method, Cancelled event variant, CannotCancelJob error, and
related tests are no longer needed.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: bodymindarts

src/entity.rs

@@ -69,7 +69,6 @@ pub enum JobEvent {
         error: String,
     },
     JobCompleted,
-    Cancelled,
     AttemptCounterReset,
 }
 
@@ -177,20 +176,12 @@ impl Job {
         serde_json::from_value(self.config.clone())
     }
 
-    /// Returns `true` once the job has emitted a `JobCompleted` or `Cancelled` event.
+    /// Returns `true` once the job has emitted a `JobCompleted` event.
     pub fn completed(&self) -> bool {
         self.events
             .iter_all()
             .rev()
-            .any(|event| matches!(event, JobEvent::JobCompleted | JobEvent::Cancelled))
-    }
-
-    /// Returns `true` if the job was cancelled.
-    pub fn cancelled(&self) -> bool {
-        self.events
-            .iter_all()
-            .rev()
-            .any(|event| matches!(event, JobEvent::Cancelled))
+            .any(|event| matches!(event, JobEvent::JobCompleted))
     }
 
     pub(crate) fn inject_tracing_parent(&self) {
@@ -236,14 +227,6 @@ impl Job {
         self.events.push(JobEvent::JobCompleted);
     }
 
-    pub(crate) fn cancel(&mut self) -> es_entity::Idempotent<()> {
-        if self.completed() {
-            return es_entity::Idempotent::AlreadyApplied;
-        }
-        self.events.push(JobEvent::Cancelled);
-        es_entity::Idempotent::Executed(())
-    }
-
     pub(super) fn schedule_retry(
         &mut self,
         error: String,
@@ -333,7 +316,6 @@ impl TryFromEvents<JobEvent> for Job {
                 JobEvent::ExecutionAborted { .. } => {}
                 JobEvent::ExecutionErrored { .. } => {}
                 JobEvent::JobCompleted => {}
-                JobEvent::Cancelled => {}
                 JobEvent::AttemptCounterReset => {}
             }
         }

src/error.rs

@@ -36,10 +36,6 @@ pub enum JobError {
     DuplicateId(Option<String>),
     #[error("JobError - DuplicateUniqueJobType: {0:?}")]
     DuplicateUniqueJobType(Option<String>),
-    #[error(
-        "JobError - CannotCancelJob: job is not in pending state (may be running or already completed)"
-    )]
-    CannotCancelJob,
     #[error("JobError - Config: {0}")]
     Config(String),
     #[error("JobError - Migration: {0}")]

src/lib.rs

@@ -220,7 +220,6 @@ mod tracker;
 
 pub mod error;
 
-use es_entity::AtomicOperation;
 use tracing::instrument;
 
 use std::sync::{Arc, Mutex};
@@ -481,35 +480,6 @@ impl Jobs {
         Ok(self.repo.find_by_id(id).await?)
     }
 
-    /// Cancel a pending job, removing it from the execution queue.
-    ///
-    /// This operation is idempotent — calling it on an already cancelled or
-    /// completed job is a no-op. If the job exists but is currently running
-    /// (not pending), returns [`JobError::CannotCancelJob`].
-    #[instrument(name = "job.cancel_job", skip(self))]
-    pub async fn cancel_job(&self, id: JobId) -> Result<(), JobError> {
-        let mut op = self.repo.begin_op_with_clock(&self.clock).await?;
-        let mut job = self.repo.find_by_id(id).await?;
-
-        if job.cancel().did_execute() {
-            let result = sqlx::query!(
-                r#"DELETE FROM job_executions WHERE id = $1 AND state = 'pending'"#,
-                id as JobId,
-            )
-            .execute(op.as_executor())
-            .await?;
-
-            if result.rows_affected() == 0 {
-                return Err(JobError::CannotCancelJob);
-            }
-
-            self.repo.update_in_op(&mut op, &mut job).await?;
-            op.commit().await?;
-        }
-
-        Ok(())
-    }
-
     /// Returns a reference to the clock used by this job service.
     pub fn clock(&self) -> &ClockHandle {
         &self.clock

tests/job.rs

@@ -638,144 +638,3 @@ async fn test_bulk_spawn_empty_batch() -> anyhow::Result<()> {
 
     Ok(())
 }
-
-#[tokio::test]
-async fn test_cancel_pending_job() -> anyhow::Result<()> {
-    let pool = helpers::init_pool().await?;
-    let config = JobSvcConfig::builder()
-        .pool(pool)
-        .build()
-        .expect("Failed to build JobsConfig");
-
-    let mut jobs = Jobs::init(config).await?;
-    let spawner = jobs.add_initializer(TestJobInitializer);
-
-    // Spawn a job scheduled far in the future so it stays pending
-    let job_id = JobId::new();
-    let schedule_at = chrono::Utc::now() + chrono::Duration::hours(24);
-    spawner
-        .spawn_at(job_id, TestJobConfig { delay_ms: 50 }, schedule_at)
-        .await?;
-
-    // Cancel the pending job
-    jobs.cancel_job(job_id).await?;
-
-    // Verify it's findable as a cancelled/completed entity
-    let found = jobs.find(job_id).await?;
-    assert!(found.completed(), "Cancelled job should be completed");
-    assert!(found.cancelled(), "Job should be marked as cancelled");
-
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_cancel_running_job_fails() -> anyhow::Result<()> {
-    let pool = helpers::init_pool().await?;
-    let config = JobSvcConfig::builder()
-        .pool(pool)
-        .build()
-        .expect("Failed to build JobsConfig");
-
-    let mut jobs = Jobs::init(config).await?;
-
-    let started = Arc::new(Mutex::new(Vec::<String>::new()));
-    let completed = Arc::new(Mutex::new(Vec::<String>::new()));
-    let release = Arc::new(Notify::new());
-
-    let spawner = jobs.add_initializer(QueueJobInitializer {
-        job_type: JobType::new("cancel-running-test"),
-        started: Arc::clone(&started),
-        completed: Arc::clone(&completed),
-        release: Arc::clone(&release),
-    });
-
-    jobs.start_poll()
-        .await
-        .expect("Failed to start job polling");
-
-    let job_id = JobId::new();
-    spawner
-        .spawn(job_id, QueueJobConfig { label: "X".into() })
-        .await?;
-
-    // Wait for the job to start running
-    let mut attempts = 0;
-    loop {
-        tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
-        if !started.lock().await.is_empty() {
-            break;
-        }
-        attempts += 1;
-        assert!(attempts < 100, "Job never started");
-    }
-
-    // Cancel on a running job should fail
-    let result = jobs.cancel_job(job_id).await;
-    assert!(
-        matches!(result, Err(JobError::CannotCancelJob)),
-        "Cancelling a running job should return JobNotPending, got err: {:?}",
-        result.err(),
-    );
-
-    // Release the job so it completes normally
-    release.notify_one();
-
-    // Wait for completion
-    let mut attempts = 0;
-    loop {
-        tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
-        let job = jobs.find(job_id).await?;
-        if job.completed() {
-            break;
-        }
-        attempts += 1;
-        assert!(attempts < 100, "Job never completed");
-    }
-
-    Ok(())
-}
-
-#[tokio::test]
-async fn test_cancel_already_completed_job_is_idempotent() -> anyhow::Result<()> {
-    let pool = helpers::init_pool().await?;
-    let config = JobSvcConfig::builder()
-        .pool(pool)
-        .build()
-        .expect("Failed to build JobsConfig");
-
-    let mut jobs = Jobs::init(config).await?;
-    let spawner = jobs.add_initializer(TestJobInitializer);
-
-    jobs.start_poll()
-        .await
-        .expect("Failed to start job polling");
-
-    let job_id = JobId::new();
-    spawner
-        .spawn(job_id, TestJobConfig { delay_ms: 10 })
-        .await?;
-
-    // Wait for the job to complete
-    let mut attempts = 0;
-    loop {
-        tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
-        let job = jobs.find(job_id).await?;
-        if job.completed() {
-            break;
-        }
-        attempts += 1;
-        assert!(attempts < 100, "Job never completed");
-    }
-
-    // Cancel on an already completed job is a no-op
-    jobs.cancel_job(job_id).await?;
-
-    let job = jobs.find(job_id).await?;
-    assert!(
-        !job.cancelled(),
-        "Completed job should not be marked cancelled"
-    );
-    assert!(job.completed(), "Job should still be completed");
-
-    Ok(())
-}