brokkr-agent::work_orders

Rust

Work Orders Module

This module handles the work order lifecycle for the Brokkr agent:

• Fetching pending work orders from the broker
• Claiming work orders for execution
• Executing work based on work type (e.g., builds)
• Reporting completion (success/failure) to the broker

Work Order Flow

1. Poll broker for pending work orders
2. Claim a work order
3. Apply YAML content (Build + WorkOrder resources)
4. Execute work type handler (e.g., create BuildRun for builds)
5. Watch for completion
6. Report result to broker

Structs

struct WorkOrderOutcomeError

pub

Derives: Debug

An outcome the agent has classified itself, so that process_single_work_order does not have to infer retryability from error text via is_error_retryable.

This exists specifically for Job and BuildRun watch outcomes. is_error_retryable classifies any message containing “timeout” as retryable, and a retry of a timed-out work order would re-dispatch the order while the original Job or BuildRun is still running in the cluster — the exact double-execution this watch is meant to prevent. Watch outcomes therefore carry their disposition explicitly rather than through wording.

Methods

is_​retryable

pub

fn is_retryable(&self) -> bool

Whether the broker should schedule a retry of this work order.

Source

    pub fn is_retryable(&self) -> bool {
        self.retryable
    }

Functions

fn process_​pending_​work_​orders

pub

async fn process_pending_work_orders(
    config: &Settings,
    http_client: &BrokkrClient,
    k8s_client: &K8sClient,
    agent: &Agent,
) -> Result<usize, Box<dyn std::error::Error + Send + Sync>>

Processes pending work orders for the agent.

This function:

1. Fetches pending work orders from the broker
2. Claims the first available work order
3. Executes the work based on work type
4. Reports completion to the broker

Parameters:

Name	Type	Description
config	&Settings	Application settings
http_client	&BrokkrClient	HTTP client for broker communication
k8s_client	&K8sClient	Kubernetes client for resource operations
agent	&Agent	Agent details

Returns:

Number of work orders processed

Source

pub async fn process_pending_work_orders(
    config: &Settings,
    http_client: &BrokkrClient,
    k8s_client: &K8sClient,
    agent: &Agent,
) -> Result<usize, Box<dyn std::error::Error + Send + Sync>> {
    // Fetch pending work orders
    let pending = broker::fetch_pending_work_orders(config, http_client, agent, None).await?;

    if pending.is_empty() {
        trace!("No pending work orders for agent {}", agent.name);
        return Ok(0);
    }

    info!(
        "Found {} pending work orders for agent {}",
        pending.len(),
        agent.name
    );

    let mut processed = 0;

    // Process one work order at a time
    // In the future, we could parallelize this based on work type
    for work_order in pending.iter().take(1) {
        match process_single_work_order(config, http_client, k8s_client, agent, work_order).await {
            Ok(_) => {
                processed += 1;
                info!(
                    "Successfully processed work order {} (type: {})",
                    work_order.id, work_order.work_type
                );
            }
            Err(e) => {
                error!(
                    "Failed to process work order {} (type: {}): {}",
                    work_order.id, work_order.work_type, e
                );
                // Continue with next work order instead of failing completely
            }
        }
    }

    Ok(processed)
}

fn watch_​job_​completion

pub

async fn watch_job_completion(
    k8s_client: &K8sClient,
    namespace: &str,
    name: &str,
    deadline: Instant,
) -> Result<(), WorkOrderOutcomeError>

Watches an applied batch/v1 Job until it reaches a terminal state or the watch budget runs out.

deadline is the absolute instant derived from the work order’s claim timeout by job_watch_budget. Status is read before the deadline is checked, so a Job that is already finished is reported correctly even when the budget is zero.

Transient API read failures do not end the watch — the agent keeps polling and, if the deadline arrives with the last read still failing, says so in the message rather than presenting it as a plain timeout.

Returns a WorkOrderOutcomeError whose message distinguishes the three non-success cases: the Job ran and failed, the Job was still running when the budget ran out, or the Job vanished.

Source

pub async fn watch_job_completion(
    k8s_client: &K8sClient,
    namespace: &str,
    name: &str,
    deadline: Instant,
) -> Result<(), WorkOrderOutcomeError> {
    let api: Api<Job> = Api::namespaced(k8s_client.clone(), namespace);
    let poll_interval = Duration::from_secs(JOB_STATUS_POLL_INTERVAL_SECS);

    loop {
        // `None` once the read succeeded, so a timeout report only mentions an
        // API problem when the *most recent* read is the one that failed.
        let last_read_error: Option<String> = match api.get_opt(name).await {
            Ok(Some(job)) => {
                let status = job.status.unwrap_or_default();
                match interpret_job_status(&status) {
                    JobOutcome::Succeeded => {
                        info!("Job {}/{} completed successfully", namespace, name);
                        return Ok(());
                    }
                    JobOutcome::Failed(detail) => {
                        error!("Job {}/{} failed: {}", namespace, name, detail);
                        return Err(WorkOrderOutcomeError::non_retryable(format!(
                            "Job {namespace}/{name} ran and failed: {detail}"
                        )));
                    }
                    JobOutcome::Running => {
                        debug!(
                            "Job {}/{} still running (active={:?}, succeeded={:?}, failed={:?})",
                            namespace, name, status.active, status.succeeded, status.failed
                        );
                        None
                    }
                }
            }
            Ok(None) => {
                return Err(WorkOrderOutcomeError::non_retryable(format!(
                    "Job {namespace}/{name} was applied but no longer exists; it was deleted \
                     before reaching a terminal state, so its outcome is unknown"
                )));
            }
            Err(e) => {
                warn!(
                    "Failed to read status of Job {}/{}, will retry: {}",
                    namespace, name, e
                );
                Some(e.to_string())
            }
        };

        let now = Instant::now();
        if now >= deadline {
            let suffix = match &last_read_error {
                Some(e) => format!(" The last status read also failed: {e}."),
                None => String::new(),
            };
            return Err(WorkOrderOutcomeError::non_retryable(format!(
                "Job {namespace}/{name} did not finish within the watch window and was still \
                 running when the agent stopped watching. The window is bounded below the work \
                 order's claim_timeout_seconds so the order cannot be reclaimed and re-run while \
                 this Job is still going. The Job was NOT cancelled and its final outcome is \
                 unknown to Brokkr — this is not a report that the Job failed.{suffix}"
            )));
        }

        // Never sleep past the deadline.
        let remaining = deadline.saturating_duration_since(now);
        tokio::time::sleep(poll_interval.min(remaining)).await;
    }
}

fn execute_​custom_​work_​order

pub

async fn execute_custom_work_order(
    k8s_client: &K8sClient,
    agent: &Agent,
    work_order: &WorkOrder,
    watch_deadline: Instant,
) -> Result<Option<String>, Box<dyn std::error::Error + Send + Sync>>

Executes a custom work order by applying YAML resources to the cluster.

After the apply, every batch/v1 Job among the applied objects is watched to a terminal state (see watch_job_completion); the work order succeeds only if all of them report Complete. Objects of any other kind are applied only — they are not monitored for completion, and the result message says so explicitly. Ongoing health of non-Job resources is the deployment-object reconciler’s concern, not the work-order path’s.

watch_deadline bounds the total time spent watching, derived from the work order’s claim_timeout_seconds by job_watch_budget.

Source

pub async fn execute_custom_work_order(
    k8s_client: &K8sClient,
    agent: &Agent,
    work_order: &WorkOrder,
    watch_deadline: Instant,
) -> Result<Option<String>, Box<dyn std::error::Error + Send + Sync>> {
    use kube::api::PatchParams;

    info!(
        "Executing custom work order {} for agent {}",
        work_order.id, agent.name
    );

    // Parse the YAML content
    let yaml_docs = crate::utils::multidoc_deserialize(&work_order.yaml_content)
        .map_err(|e| format!("failed to parse work order yaml: {e}"))?;

    if yaml_docs.is_empty() {
        return Err("Work order YAML content is empty".into());
    }

    // Convert YAML docs to DynamicObjects
    let mut objects: Vec<DynamicObject> = Vec::new();
    for yaml_doc in &yaml_docs {
        // Skip null documents
        if yaml_doc.is_null() {
            continue;
        }

        let object: DynamicObject = serde_yaml::from_value(yaml_doc.clone())?;
        let gvk = object
            .types
            .as_ref()
            .ok_or("Object missing type metadata")?;
        debug!(
            "Parsed {} {}/{}",
            gvk.kind,
            object.metadata.namespace.as_deref().unwrap_or("cluster"),
            object.metadata.name.as_deref().unwrap_or("unnamed")
        );
        objects.push(object);
    }

    if objects.is_empty() {
        return Err("No valid Kubernetes objects found in YAML".into());
    }

    info!(
        "Applying {} resource(s) from custom work order {}",
        objects.len(),
        work_order.id
    );

    // Apply all resources using server-side apply
    let patch_params = PatchParams::apply("brokkr-agent").force();
    crate::k8s::api::apply_k8s_objects(&objects, k8s_client.clone(), patch_params)
        .await
        .map_err(|e| format!("failed to apply work order objects: {e}"))?;

    // Applying is not finishing. Watch every batch/v1 Job to a terminal state;
    // every other kind stays apply-only.
    let (jobs, others): (Vec<&DynamicObject>, Vec<&DynamicObject>) =
        objects.iter().partition(|o| is_batch_v1_job(o));

    let unmonitored = describe_unmonitored_kinds(&others);

    if jobs.is_empty() {
        return Ok(Some(format!(
            "Applied {} resource(s). No batch/v1 Job present, so nothing was watched to \
             completion: {} applied only, not monitored. Only batch/v1 Jobs have a terminal \
             state the agent waits on.",
            objects.len(),
            unmonitored
        )));
    }

    let mut watched: Vec<String> = Vec::with_capacity(jobs.len());
    for job in &jobs {
        let namespace = job.metadata.namespace.as_deref().unwrap_or("default");
        let Some(name) = job.metadata.name.as_deref() else {
            return Err(Box::new(WorkOrderOutcomeError::non_retryable(
                "A batch/v1 Job in this work order has no metadata.name, so it cannot be watched \
                 to completion. Give the Job an explicit name (generateName is not supported \
                 here)."
                    .to_string(),
            )));
        };

        info!(
            "Watching Job {}/{} from custom work order {} until it reaches a terminal state",
            namespace, name, work_order.id
        );
        watch_job_completion(k8s_client, namespace, name, watch_deadline)
            .await
            .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send + Sync>)?;
        watched.push(format!("{namespace}/{name}"));
    }

    let others_note = if others.is_empty() {
        String::new()
    } else {
        format!(" {unmonitored} applied only, not monitored.")
    };

    Ok(Some(format!(
        "Applied {} resource(s). Job(s) {} completed successfully.{}",
        objects.len(),
        watched.join(", "),
        others_note
    )))
}