args-cli 0.1.1

Command-line interface for args
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

use std::sync::Arc;

use anyhow::Context;

use crate::{
    client::ArgsClient,
    config::RunnerConfig,
    executor::{Executor, local::LocalExecutor},
};

pub async fn run(config: RunnerConfig) -> anyhow::Result<()> {
    if config.runner_token.is_none() {
        eprintln!("Error: runner is not installed. Please run `args runner install` first.");
        return Ok(());
    }

    let config = Arc::new(config);
    let runner_client = Arc::new(ArgsClient::from_runner_config(&config));
    let semaphore = Arc::new(tokio::sync::Semaphore::new(config.num_executors as usize));

    loop {
        let permit = Arc::clone(&semaphore)
            .acquire_owned()
            .await
            .context("semaphore closed")?;

        let task = match runner_client.poll_task(()).await {
            Ok(Some(task)) => task,
            Ok(None) => {
                drop(permit);
                continue;
            }
            Err(e) => {
                eprintln!("Error polling task: {:#?}", e);
                drop(permit);
                continue;
            }
        };

        let task_client =
            Arc::new(ArgsClient::from_runner_config(&config).with_jwt(task.token.clone()));
        let config = Arc::clone(&config);

        tokio::spawn(async move {
            let _permit = permit;
            if let Err(e) = task_client.update_task(task.id, "running").await {
                eprintln!("Failed to mark task {} as running: {}", task.id, e);
                return;
            }

            let executor = match LocalExecutor::initialize(&config, &task).await {
                Ok(e) => e,
                Err(e) => {
                    eprintln!("Failed to initialize executor: {}", e);
                    return;
                }
            };

            let task_id = executor.task.id;
            let result = executor.execute().await;

            let final_status = match result {
                Ok(()) => "success",
                Err(ref e) => {
                    eprintln!("Task {} failed: {}", task_id, e);
                    "failure"
                }
            };

            if let Err(e) = task_client.update_task(task_id, final_status).await {
                eprintln!("Failed to mark task {} as {}: {}", task_id, final_status, e);
            }

            if let Err(e) = executor.cleanup().await {
                eprintln!("Failed to clean up task {}: {}", task_id, e);
            }
        });
    }
}