Checkpoints, so a handed-back job resumes (refs andymccall/oxygenflow#77) e4203e141e90

- **Checkpoints, so a handed-back job resumes (oxygenflow#77):** `POST /api/v1/flow/jobs/{id}/checkpoint` keeps the directories a job declares with `checkpoint "/src/…"` as its runner hands it back; the next attempt's claim names the blob (`checkpoint`) and its runner fetches it (`GET`) and restores it before running. A checkpoint is kept beside the job's artifacts but is none of them (no receipt names it). Pins OxygenFlow's main.

            "/flow/jobs/{id}/checkpoint",

            post(crate::flow::save_checkpoint)

                .get(crate::flow::checkpoint)

                .layer(axum::extract::DefaultBodyLimit::max(crate::flow::MAX_ARTIFACT_BYTES + (1 << 20))),

        )

        .route(

/// Job `job_id`'s artifacts, by name; not its [`CHECKPOINT`], which no

/// receipt names.

    sqlx::query_as("SELECT * FROM flow_artifacts WHERE job_id = $1 AND name NOT LIKE '.%' ORDER BY name").bind(job_id).fetch_all(pool).await

/// The name a job's checkpoint is kept under (oxygenflow#77), beside its

/// artifacts: no pipeline can declare it.

pub const CHECKPOINT: &str = ".checkpoint";

/// The checkpoint an attempt of job `job_id` saved as its runner handed it

/// back (oxygenflow#77), if one did.

pub async fn checkpoint(pool: &Pool, job_id: i64) -> sqlx::Result<Option<Artifact>> {

    sqlx::query_as("SELECT * FROM flow_artifacts WHERE job_id = $1 AND name = $2").bind(job_id).bind(CHECKPOINT).fetch_optional(pool).await

}

        // What an attempt handed back saved, to resume from (oxygenflow#77).

        checkpoint: if job.handbacks > 0 { fdb::checkpoint(&app.db, job.id).await?.map(|a| a.blob) } else { None },

/// `POST /api/v1/flow/jobs/{id}/checkpoint` (oxygenflow#77): a runner

/// handing a job back keeps its checkpoint directories, packed; kept as a

/// blob of the job's repository, for the next attempt's claim to name.

pub async fn save_checkpoint(

    State(app): State<Shared>,

    headers: HeaderMap,

    Path(id): Path<i64>,

    body: axum::body::Bytes,

) -> Result<Response, ApiError> {

    let runner = runner_auth(&app, &headers).await?;

    let job = running_job(&app, id).await?;

    if runner.source == "registered" && job.runner != runner.name {

        return Err(ApiError(StatusCode::FORBIDDEN, "this job is held by another runner".into()));

    }

    let spec: flow_core::Job =

        serde_json::from_str(&job.spec).map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("stored job: {e}")))?;

    if spec.checkpoints.is_empty() {

        return Err(ApiError(StatusCode::BAD_REQUEST, format!("job {} declares no checkpoint", job.name)));

    }

    let repo = job_repo(&app, &job).await?;

    let content = blake3::hash(&body).to_hex().to_string();

    let size = body.len() as i64;

    let blob = app

        .repos

        .write(&repo, move |r, _| Ok(r.store_blob(&body)?))

        .await

        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("keep the checkpoint: {e}")))?;

    let a = fdb::Artifact {

        job_id: job.id,

        name: fdb::CHECKPOINT.into(),

        path: spec.checkpoints.join(" "),

        blob: blob.to_hex(),

        size,

        content,

        created_ms: db::now_ms(),

    };

    fdb::add_artifact(&app.db, &a).await?;

    Ok(Json(serde_json::json!({ "blob": a.blob })).into_response())

}

/// `GET /api/v1/flow/jobs/{id}/checkpoint` (oxygenflow#77): the checkpoint

/// a handed-back attempt saved, for the attempt now running it.

pub async fn checkpoint(State(app): State<Shared>, headers: HeaderMap, Path(id): Path<i64>) -> Result<Response, ApiError> {

    let runner = runner_auth(&app, &headers).await?;

    let job = running_job(&app, id).await?;

    if runner.source == "registered" && job.runner != runner.name {

        return Err(ApiError(StatusCode::FORBIDDEN, "this job is held by another runner".into()));

    }

    let a = fdb::checkpoint(&app.db, job.id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "this job has no checkpoint".into()))?;

    let repo = job_repo(&app, &job).await?;

    let blob = oxy_core::hash::ObjectRef::from_hex(&a.blob)

        .ok_or_else(|| ApiError(StatusCode::INTERNAL_SERVER_ERROR, "a kept checkpoint has a bad blob".into()))?;

    let bytes = app

        .repos

        .read(&repo, move |r, _| Ok(r.read_blob(&blob)?))

        .await

        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("read the checkpoint: {e}")))?;

    Ok(([(axum::http::header::CONTENT_TYPE, "application/octet-stream")], bytes).into_response())

}

/// The repository job `job` runs for.

async fn job_repo(app: &Shared, job: &fdb::Job) -> Result<RepoName, ApiError> {

    let run = fdb::run_by_id(&app.db, job.run_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such run".into()))?;

    let row =

        db::repos::get_by_id(&app.db, run.repo_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such repository".into()))?;

    RepoName::new(&row.owner, &row.name).ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "bad repository name".into()))

}

    let bulk = "job \"bulk\" {\n    sandbox \"alpine\"\n    priority \"low\"\n    checkpoint \"/src/state\"\n    exec \"true\"\n}\n";

    // It keeps its checkpoint first (oxygenflow#77), a job's own only.

    let (st, body) = call(&h.app, "POST", &format!("/api/v1/flow/jobs/{lint}/checkpoint"), Some("runner-two"), Some("x".into())).await;

    assert_eq!(st, StatusCode::BAD_REQUEST, "lint declares no checkpoint: {body}");

    let (st, body) =

        call(&h.app, "POST", &format!("/api/v1/flow/jobs/{low}/checkpoint"), Some("runner-three"), Some("step 32".into())).await;

    assert_eq!(st, StatusCode::OK, "{body}");

    let saved = serde_json::from_str::<serde_json::Value>(&body).unwrap()["blob"].as_str().unwrap().to_string();

    let body = serde_json::json!({ "runner": "phone", "runtimes": ["microvm"] }).to_string();

    let (st, body) = call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-three"), Some(body)).await;

    assert_eq!(st, StatusCode::OK, "{body}");

    let again: serde_json::Value = serde_json::from_str(&body).unwrap();

    assert_eq!((again["job_id"].as_i64(), again["checkpoint"].as_str()), (Some(low), Some(saved.as_str())), "it resumes: {again}");

    let (st, bytes) = call(&h.app, "GET", &format!("/api/v1/flow/jobs/{low}/checkpoint"), Some("runner-three"), None).await;

    assert_eq!((st, bytes.as_str()), (StatusCode::OK, "step 32"));

    assert!(db::flow::artifacts(&h.app.db, low).await.unwrap().is_empty(), "a checkpoint is no artifact of the job");