cjw/Oxicloud
1
0
Fork 0
Code Issues Pull Requests Actions 4 Packages Projects Releases Wiki Activity

feat(recoverable jobs): clarify life cycle pause vs cancel

Browse Source 
a job can be paused/resumed
    a job as an exclusibity by it's name
    if you want to run another job with same name:
    either cancel the first one, or wait of it's terminaison

    pause does not permit to run the other job, this can create
    race conditions
This commit is contained in:
Edouard Vanbelle
2026-08-02 16:55:07 +02:00
parent 5eec0fb36e
commit 8b0fb03b5c
7 changed files with 511 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/infrastructure/scheduler/pg_job_store.rs
+115
View File
@@ -345,6 +345,45 @@ impl JobStore for PgJobStore {
.map_err(|e| map_sqlx_err("mark_failed", e))?;
Ok(())
}
async fn mark_cancelled(&self, cursor: Option<Vec<u8>>) -> Result<(), DomainError> {
// Mirror of `mark_paused`'s two-branch cursor handling: preserve
// the last-known cursor for post-mortem inspection (an operator
// can see how far the abandoned run got) even though nothing
// will resume it.
if let Some(c) = cursor {
sqlx::query(
r#"
UPDATE jobs.recoverable_runs
SET status = 'Cancelled',
cursor = $2,
completed_at = NOW(),
last_progress_at = NOW()
WHERE id = $1
"#,
)
.bind(self.run_id)
.bind(&c[..])
.execute(self.pool.as_ref())
.await
.map_err(|e| map_sqlx_err("mark_cancelled", e))?;
} else {
sqlx::query(
r#"
UPDATE jobs.recoverable_runs
SET status = 'Cancelled',
completed_at = NOW(),
last_progress_at = NOW()
WHERE id = $1
"#,
)
.bind(self.run_id)
.execute(self.pool.as_ref())
.await
.map_err(|e| map_sqlx_err("mark_cancelled", e))?;
}
Ok(())
}
}
// ─── PgJobStoreProvider — registry-level ops ────────────────────────────────
@@ -558,6 +597,82 @@ impl JobStoreProvider for PgJobStoreProvider {
.map_err(|e| map_sqlx_err("request_cancel", e))?;
Ok(flipped.map(|(id,)| id))
}
async fn request_terminal_cancel(&self, job_name: &str) -> Result<Option<Uuid>, DomainError> {
// Latest non-terminal row for this job. Order by started_at DESC
// + LIMIT 1 defends against partial-index churn during retries.
let row: Option<(Uuid, String)> = sqlx::query_as(
r#"
SELECT id, status FROM jobs.recoverable_runs
WHERE job_name = $1
AND status IN ('Running', 'CancelRequested', 'Paused')
ORDER BY started_at DESC
LIMIT 1
"#,
)
.bind(job_name)
.fetch_optional(self.pool.as_ref())
.await
.map_err(|e| map_sqlx_err("request_terminal_cancel.select", e))?;
let Some((id, status)) = row else {
return Ok(None);
};
match status.as_str() {
"Paused" => {
// Direct DB flip — no handler is running to observe
// the intent flag, so we transition immediately.
sqlx::query(
r#"
UPDATE jobs.recoverable_runs
SET status = 'Cancelled',
completed_at = NOW(),
last_progress_at = NOW()
WHERE id = $1
AND status = 'Paused'
"#,
)
.bind(id)
.execute(self.pool.as_ref())
.await
.map_err(|e| map_sqlx_err("request_terminal_cancel.paused_flip", e))?;
Ok(Some(id))
}
"Running" | "CancelRequested" => {
// Stamp intent + flip to CancelRequested in one statement.
// The handler's next `store.status()` poll observes
// CancelRequested, returns `RunOutcome::Paused` at the
// next boundary; the engine wrap reads the intent and
// calls `mark_cancelled` instead of `mark_paused`.
//
// If the row was already CancelRequested (admin clicked
// Pause first, then Cancel), the status update is a
// no-op but the intent flag stamps — the engine wrap
// upgrades the pending Paused into Cancelled at yield
// time.
sqlx::query(
r#"
UPDATE jobs.recoverable_runs
SET status = 'CancelRequested',
params = jsonb_set(COALESCE(params, '{}'::jsonb),
'{cancel_intent}',
'"terminate"'::jsonb),
last_progress_at = NOW()
WHERE id = $1
"#,
)
.bind(id)
.execute(self.pool.as_ref())
.await
.map_err(|e| map_sqlx_err("request_terminal_cancel.running_flip", e))?;
Ok(Some(id))
}
other => Err(DomainError::internal_error(
"JobStore",
format!("request_terminal_cancel: unexpected status `{other}`"),
)),
}
}
}
// ─── Shared row → RunSummary decoder ────────────────────────────────────────
src/infrastructure/scheduler/recoverable.rs
+134 -15
View File
@@ -54,13 +54,21 @@ use super::types::{JobOutcome, JobRunArgs};
/// Mirror of the `TEXT` values allowed in `jobs.recoverable_runs.status`.
///
/// Terminal set = `{Completed, Failed}`. Non-terminal set (the one the
/// exclusivity partial unique index scopes) =
/// Terminal set = `{Completed, Failed, Cancelled}`. Non-terminal set
/// (the one the exclusivity partial unique index scopes) =
/// `{Running, Paused, CancelRequested}`.
///
/// `CancelRequested` IS non-terminal — the run is still shutting down.
/// A second trigger arriving during cancel MUST NOT spawn a parallel
/// run; the trigger endpoint returns the surviving row instead.
///
/// `Cancelled` IS terminal — admin explicitly abandoned the run. Distinct
/// from `Failed` because it's user-driven, not a handler error. Distinct
/// from `Paused` because it's not resumable. Runs land in `Cancelled` via
/// two paths: (1) admin cancel on a Running row (sets
/// `params.cancel_intent = "terminate"` alongside the CancelRequested
/// flip; engine post-processes handler's Paused return → Cancelled), or
/// (2) admin cancel on an already-Paused row (direct DB flip).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
pub enum RunStatus {
Running,
@@ -68,6 +76,7 @@ pub enum RunStatus {
CancelRequested,
Completed,
Failed,
Cancelled,
}
impl RunStatus {
@@ -79,6 +88,7 @@ impl RunStatus {
RunStatus::CancelRequested => "CancelRequested",
RunStatus::Completed => "Completed",
RunStatus::Failed => "Failed",
RunStatus::Cancelled => "Cancelled",
}
}
@@ -91,6 +101,7 @@ impl RunStatus {
"CancelRequested" => Some(RunStatus::CancelRequested),
"Completed" => Some(RunStatus::Completed),
"Failed" => Some(RunStatus::Failed),
"Cancelled" => Some(RunStatus::Cancelled),
_ => None,
}
}
@@ -105,6 +116,14 @@ impl RunStatus {
}
}
/// Value written to `params.cancel_intent` to tell the engine's
/// terminal-write wrap how to interpret a subsequent
/// [`RunOutcome::Paused`] return. Absent → treat as ordinary pause
/// (write `Paused`). Present with this value → the admin asked to
/// abandon, not just yield, so write `Cancelled` instead.
pub const CANCEL_INTENT_PARAM: &str = "cancel_intent";
pub const CANCEL_INTENT_TERMINATE: &str = "terminate";
// ─── Run outcome (handler → engine) ─────────────────────────────────────────
/// What a [`RecoverableJobHandler`] returns from `run_resumable`.
@@ -395,6 +414,15 @@ pub trait JobStore: Send + Sync {
/// Engine-only. Called by [`run_or_resume`] on
/// [`RunOutcome::Failed`]. Handler code MUST NOT call this.
async fn mark_failed(&self, message: &str) -> Result<(), DomainError>;
/// Engine-only. Called by [`run_or_resume`] when the handler
/// returns [`RunOutcome::Paused`] AND
/// `params.cancel_intent = "terminate"` — the admin asked to
/// abandon the run, not just yield. Writes `status = 'Cancelled'`
/// + `completed_at = NOW()`. Preserves the cursor for post-mortem
/// (an operator can see how far it got before being killed).
/// Handler code MUST NOT call this.
async fn mark_cancelled(&self, cursor: Option<Vec<u8>>) -> Result<(), DomainError>;
}
/// Registry-level operations on `jobs.recoverable_runs` — NOT bound
@@ -451,6 +479,25 @@ pub trait JobStoreProvider: Send + Sync {
/// completes naturally.
async fn request_cancel(&self, job_name: &str) -> Result<Option<Uuid>, DomainError>;
/// Request TERMINAL cancellation — admin abandons the run rather
/// than yielding it for later resume. Two paths depending on the
/// current row's status:
///
/// - **`Running` / `CancelRequested`** — same DB flip as
/// [`Self::request_cancel`] (Running → CancelRequested) BUT
/// also stamps `params.cancel_intent = "terminate"`. When the
/// handler yields and the engine wraps `RunOutcome::Paused`, it
/// reads the intent and calls
/// [`JobStore::mark_cancelled`] instead of `mark_paused`.
/// - **`Paused`** — no handler is running, so the engine wrap
/// never fires. Direct DB flip `Paused → Cancelled +
/// completed_at = NOW()`.
/// - **Terminal or absent** — no-op (`Ok(None)`).
///
/// Returns the affected run's id when any transition happened,
/// `None` otherwise.
async fn request_terminal_cancel(&self, job_name: &str) -> Result<Option<Uuid>, DomainError>;
/// Findings for a specific run, newest-last, paginated.
/// Powers `GET /api/admin/jobs/{name}/runs/{id}/findings`.
/// `limit` caps rows; the API layer clamps it too. `offset` is
@@ -745,19 +792,54 @@ pub async fn run_or_resume(
)
}
RunOutcome::Paused { cursor } => {
// Read the intent stamped by `/api/admin/jobs/{name}/cancel`
// (terminal cancel path). Absent → ordinary pause. Present
// with `terminate` → admin asked to abandon; write
// Cancelled instead of Paused. Any read error falls
// through to Paused — errs on preserving-progress side.
let terminate = store
.get_string_param(CANCEL_INTENT_PARAM)
.await
.ok()
.flatten()
.as_deref()
== Some(CANCEL_INTENT_TERMINATE);
let cursor_hex = hex::encode(&cursor);
log_terminal_write_err("mark_paused", run_id, store.mark_paused(Some(cursor)).await);
JobOutcome::ok_with(
stats.finding_count,
serde_json::json!({
"paused": true,
"run_id": run_id.to_string(),
"cursor_hex": cursor_hex,
"finding_count": stats.finding_count,
"scanned_count": stats.scanned_count,
"severity_counts": stats.by_severity,
}),
)
if terminate {
log_terminal_write_err(
"mark_cancelled",
run_id,
store.mark_cancelled(Some(cursor)).await,
);
JobOutcome::ok_with(
stats.finding_count,
serde_json::json!({
"cancelled": true,
"run_id": run_id.to_string(),
"cursor_hex": cursor_hex,
"finding_count": stats.finding_count,
"scanned_count": stats.scanned_count,
"severity_counts": stats.by_severity,
}),
)
} else {
log_terminal_write_err(
"mark_paused",
run_id,
store.mark_paused(Some(cursor)).await,
);
JobOutcome::ok_with(
stats.finding_count,
serde_json::json!({
"paused": true,
"run_id": run_id.to_string(),
"cursor_hex": cursor_hex,
"finding_count": stats.finding_count,
"scanned_count": stats.scanned_count,
"severity_counts": stats.by_severity,
}),
)
}
}
RunOutcome::Failed { message } => {
log_terminal_write_err("mark_failed", run_id, store.mark_failed(&message).await);
@@ -1060,6 +1142,14 @@ mod tests {
s.error_message = Some(message.to_string());
Ok(())
}
async fn mark_cancelled(&self, cursor: Option<Vec<u8>>) -> Result<(), DomainError> {
let mut s = self.state.lock().unwrap();
s.status = RunStatus::Cancelled;
if let Some(c) = cursor {
s.cursor = Some(c);
}
Ok(())
}
}
// ─── In-memory JobStoreProvider ────────────────────────────────────────
@@ -1291,7 +1381,10 @@ mod tests {
let before = stores.len();
stores.retain(|s| {
let state = s.state.lock().unwrap();
!matches!(state.status, RunStatus::Completed | RunStatus::Failed)
!matches!(
state.status,
RunStatus::Completed | RunStatus::Failed | RunStatus::Cancelled
)
});
Ok((before - stores.len()) as u64)
}
@@ -1307,6 +1400,32 @@ mod tests {
}
Ok(None)
}
async fn request_terminal_cancel(
&self,
_job_name: &str,
) -> Result<Option<Uuid>, DomainError> {
let stores = self.stores.lock().unwrap();
if let Some(s) = stores.last() {
let mut state = s.state.lock().unwrap();
match state.status {
RunStatus::Paused => {
state.status = RunStatus::Cancelled;
return Ok(Some(s.run_id));
}
RunStatus::Running | RunStatus::CancelRequested => {
state.status = RunStatus::CancelRequested;
state.string_params.insert(
CANCEL_INTENT_PARAM.to_string(),
CANCEL_INTENT_TERMINATE.to_string(),
);
return Ok(Some(s.run_id));
}
_ => {}
}
}
Ok(None)
}
}
// ─── Handlers ──────────────────────────────────────────────────────────
src/interfaces/api/handlers/admin_handler.rs
+78 -4
View File
@@ -177,6 +177,7 @@ pub fn admin_routes() -> Router<Arc<AppState>> {
.route("/jobs", get(list_jobs))
.route("/jobs/{name}/trigger", post(trigger_job))
.route("/jobs/{name}/cancel", post(cancel_job))
.route("/jobs/{name}/pause", post(pause_job))
.route("/jobs/{name}/runs", get(list_job_runs))
.route("/jobs/{name}/runs/{id}", get(get_job_run))
.route(
@@ -799,6 +800,10 @@ fn run_to_migration_dto(
RunStatus::CancelRequested => "paused",
RunStatus::Completed => "completed",
RunStatus::Failed => "failed",
// Cancelled is user-abandoned but terminal — same visual as
// failed for the migration status endpoint (both mean "not
// going to finish, look at findings/logs to know why").
RunStatus::Cancelled => "cancelled",
}
.to_string();
@@ -2559,16 +2564,28 @@ pub async fn cancel_job(
target: "audit",
event = "job.cancel_requested",
job = %name,
"👮🏻‍♂️ Admin requested cancel for job {}",
"👮🏻‍♂️ Admin requested TERMINAL cancel for job {}",
name,
);
match state.core.job_store_provider.request_cancel(&name).await {
// Terminal semantics: stamps `params.cancel_intent = "terminate"`
// when a Running / CancelRequested row is present so the engine
// upgrades the handler's yield to `Cancelled` instead of `Paused`.
// When the current row is `Paused` (no handler running), does a
// direct DB flip Paused → Cancelled. See
// `PgJobStoreProvider::request_terminal_cancel`.
match state
.core
.job_store_provider
.request_terminal_cancel(&name)
.await
{
Ok(Some(run_id)) => (
StatusCode::OK,
Json(serde_json::json!({
"cancelled": true,
"run_id": run_id.to_string(),
"status": "CancelRequested",
"note": "Running row → will land in Cancelled at next batch boundary; \
Paused row → flipped to Cancelled immediately.",
})),
)
.into_response(),
@@ -2576,7 +2593,7 @@ pub async fn cancel_job(
StatusCode::OK,
Json(serde_json::json!({
"cancelled": false,
"reason": "no running run for this job",
"reason": "no non-terminal run for this job",
})),
)
.into_response(),
@@ -2584,6 +2601,63 @@ pub async fn cancel_job(
}
}
/// `POST /api/admin/jobs/{name}/pause` — cooperative PAUSE of the
/// currently-running recoverable run for `{name}`.
///
/// Same DB mechanism as the old cancel (Running → CancelRequested,
/// handler yields to Paused), but no `cancel_intent` stamp so the
/// engine writes `Paused`. Use this to interrupt a long-running
/// job and resume it later; use `/cancel` to abandon it terminally.
///
/// Idempotent: if the row is already Paused, returns 200 with
/// `paused: false, reason: "already_paused"`.
#[utoipa::path(
post,
path = "/api/admin/jobs/{name}/pause",
params(("name" = String, Path, description = "Registered job name")),
responses(
(status = 200, description = "Pause signalled (or no-op if nothing was running)"),
(status = 401, description = "Unauthorized"),
(status = 403, description = "Admin required"),
(status = 500, description = "DB error"),
),
security(("bearerAuth" = [])),
tag = "admin"
)]
pub async fn pause_job(
State(state): State<Arc<AppState>>,
axum::extract::Path(name): axum::extract::Path<String>,
) -> impl IntoResponse {
use crate::infrastructure::scheduler::JobStoreProvider as _;
tracing::info!(
target: "audit",
event = "job.pause_requested",
job = %name,
"👮🏻‍♂️ Admin requested pause for job {}",
name,
);
match state.core.job_store_provider.request_cancel(&name).await {
Ok(Some(run_id)) => (
StatusCode::OK,
Json(serde_json::json!({
"paused": true,
"run_id": run_id.to_string(),
"note": "Handler will yield at the next batch boundary; row will land in Paused.",
})),
)
.into_response(),
Ok(None) => (
StatusCode::OK,
Json(serde_json::json!({
"paused": false,
"reason": "no running run for this job",
})),
)
.into_response(),
Err(e) => AppError::internal_error(format!("pause failed: {e}")).into_response(),
}
}
/// Query parameters for `GET /api/admin/jobs/{name}/runs`.
#[derive(serde::Deserialize)]
pub struct ListRunsQuery {