aion-server 0.31.0

//! The boot-time orphan pass: an auto record whose staged workflow type no
//! deployed package carries is withdrawn by name; every other record stays.

use std::collections::BTreeSet;
use std::sync::Arc;

use aion_store::{
    DeployedBinaryIdentity, DesiredState, InMemoryStore, NewWorkerDeployment, WorkerArtifactRef,
    WorkerDeployment,
};
use chrono::Utc;

use super::withdraw_orphaned;
use crate::ServerState;
use crate::worker::auto_provision::AutoWorkerDecision;
use crate::worker::auto_provision::test_support::{liminal_outbox, state_over};

type TestResult = Result<(), Box<dyn std::error::Error>>;

struct Fixture {
    state: ServerState,
    /// The temporary home the staged documents are named under.
    home: tempfile::TempDir,
}

impl Fixture {
    fn new() -> Result<Self, Box<dyn std::error::Error>> {
        let home = tempfile::tempdir()?;
        let state = state_over(
            Arc::new(InMemoryStore::default()),
            liminal_outbox(),
            true,
            home.path(),
        )?;
        Ok(Self { state, home })
    }

    /// A record as the provisioner mints one: its argv names a staged document
    /// under `<type>@<digest>/`.
    async fn put_auto_record(
        &self,
        name: &str,
        task_queue: &str,
        workflow_type: &str,
    ) -> TestResult {
        let document = self
            .home
            .path()
            .join("workers/documents")
            .join(format!("{workflow_type}@0123abcd"))
            .join(format!("{workflow_type}.awl"));
        self.put_record(
            name,
            task_queue,
            vec![
                "worker".to_owned(),
                "agent".to_owned(),
                document.to_string_lossy().into_owned(),
                "--task-queue".to_owned(),
                task_queue.to_owned(),
            ],
        )
        .await
    }

    /// A record an operator wrote: no staged document in its argv.
    async fn put_operator_record(&self, name: &str, task_queue: &str) -> TestResult {
        self.put_record(
            name,
            task_queue,
            vec!["-c".to_owned(), "sleep 300".to_owned()],
        )
        .await
    }

    async fn put_record(&self, name: &str, task_queue: &str, verb: Vec<String>) -> TestResult {
        let record = WorkerDeployment::new(
            NewWorkerDeployment {
                name: name.to_owned(),
                artifact: WorkerArtifactRef::Builtin { verb },
                binary: DeployedBinaryIdentity {
                    version: "test".to_owned(),
                    commit: "test".to_owned(),
                    dirty: "false".to_owned(),
                    content_hash: "test".to_owned(),
                },
                namespaces: BTreeSet::from(["default".to_owned()]),
                task_queue: task_queue.to_owned(),
                node: None,
                desired: DesiredState::Stopped,
            },
            Utc::now(),
        )?;
        drop(
            self.state
                .worker_deployment_store()
                .put_worker_deployment(record)
                .await?,
        );
        Ok(())
    }

    async fn names(&self) -> Result<Vec<String>, Box<dyn std::error::Error>> {
        Ok(self
            .state
            .worker_deployment_store()
            .list_worker_deployments()
            .await?
            .deployments
            .into_iter()
            .map(|deployment| deployment.name)
            .collect())
    }

    async fn teardown(&self) {
        drop(self.state.worker_supervisor().shutdown().await);
    }
}

#[tokio::test]
async fn an_auto_record_whose_type_has_no_package_is_withdrawn_and_every_other_record_stays()
-> TestResult {
    let fixture = Fixture::new()?;
    fixture
        .put_auto_record("auto/assistant", "assistant", "assistant")
        .await?;
    fixture
        .put_auto_record("auto/demo_agent", "demo_agent", "demo_session")
        .await?;
    fixture
        .put_operator_record("ops/hand-written", "ops")
        .await?;

    let deployed = BTreeSet::from(["demo_session".to_owned()]);
    let outcomes = withdraw_orphaned(&fixture.state, &deployed).await;

    assert_eq!(outcomes.len(), 1, "{outcomes:?}");
    assert_eq!(outcomes[0].decision, AutoWorkerDecision::Retired);
    assert_eq!(outcomes[0].task_queue, "assistant");
    assert_eq!(outcomes[0].workflow_type, "assistant");
    assert!(
        outcomes[0]
            .detail
            .contains("no deployed package carries workflow type `assistant`"),
        "{}",
        outcomes[0].detail
    );
    let mut names = fixture.names().await?;
    names.sort();
    assert_eq!(
        names,
        vec!["auto/demo_agent".to_owned(), "ops/hand-written".to_owned()],
        "the record with a package and the operator's record both stay"
    );
    // The decision is on the status surface, where the console reads it.
    let report = fixture.state.worker_supervisor().report().await?;
    assert!(
        report
            .auto_provision
            .iter()
            .any(|entry| entry.task_queue == "assistant"
                && entry.decision == AutoWorkerDecision::Retired),
        "{:?}",
        report.auto_provision
    );
    fixture.teardown().await;
    Ok(())
}

#[tokio::test]
async fn a_catalogue_with_nothing_deployed_withdraws_every_auto_record_and_no_operator_record()
-> TestResult {
    // The other arm: an EMPTY catalogue that was read is a real answer — no
    // workflow is deployed, so no auto worker has anything to serve. (An
    // UNREADABLE catalogue never reaches this function; the caller keeps it
    // out, and that arm is the caller's to log.)
    let fixture = Fixture::new()?;
    fixture
        .put_auto_record("auto/assistant", "assistant", "assistant")
        .await?;
    fixture
        .put_operator_record("ops/hand-written", "ops")
        .await?;

    let outcomes = withdraw_orphaned(&fixture.state, &BTreeSet::new()).await;

    assert_eq!(outcomes.len(), 1, "{outcomes:?}");
    assert_eq!(outcomes[0].decision, AutoWorkerDecision::Retired);
    assert_eq!(fixture.names().await?, vec!["ops/hand-written".to_owned()]);
    fixture.teardown().await;
    Ok(())
}

#[tokio::test]
async fn a_type_with_a_character_the_directory_cannot_carry_still_matches_its_package() -> TestResult
{
    // The staged directory spells `my.flow` as `my_flow`; the deployed set is
    // spelled the same way by `deployed_workflow_types`, so the record is NOT
    // an orphan. A raw comparison would withdraw a live worker here.
    let fixture = Fixture::new()?;
    fixture
        .put_auto_record("auto/dotted", "dotted", "my_flow")
        .await?;
    let deployed = BTreeSet::from(["my_flow".to_owned()]);
    let outcomes = withdraw_orphaned(&fixture.state, &deployed).await;
    assert!(outcomes.is_empty(), "{outcomes:?}");
    assert_eq!(fixture.names().await?, vec!["auto/dotted".to_owned()]);
    fixture.teardown().await;
    Ok(())
}