arqen 0.1.0

Backend infrastructure for agent-ready applications
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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131

use std::sync::Arc;
use tokio::sync::watch;
use tokio::time::{Duration, sleep};
use tracing::{error, info, warn};

pub mod worker;

pub use worker::Worker;

#[derive(Debug, Clone)]
pub struct JobConfig {
    pub queue: String,
    pub worker_id: String,
    pub poll_interval: Duration,
    pub lease_seconds: u32,
    pub max_retries: u32,
}

impl Default for JobConfig {
    fn default() -> Self {
        Self {
            queue: "default".to_string(),
            worker_id: uuid::Uuid::new_v4().to_string(),
            poll_interval: Duration::from_secs(1),
            lease_seconds: 30,
            max_retries: 3,
        }
    }
}

pub struct JobWorker {
    config: JobConfig,
    thingd: Arc<dyn crate::thingd::ThingdBackend>,
    handler: Box<dyn JobHandler>,
    shutdown_rx: watch::Receiver<bool>,
}

#[async_trait::async_trait]
pub trait JobHandler: Send + Sync {
    async fn handle(&self, payload: serde_json::Value) -> Result<(), crate::core::AppError>;
}

impl JobWorker {
    pub fn new(
        config: JobConfig,
        thingd: Arc<dyn crate::thingd::ThingdBackend>,
        handler: Box<dyn JobHandler>,
        shutdown_rx: watch::Receiver<bool>,
    ) -> Self {
        Self {
            config,
            thingd,
            handler,
            shutdown_rx,
        }
    }

    pub async fn run(&mut self) {
        info!(
            "Starting worker {} for queue {}",
            self.config.worker_id, self.config.queue
        );

        loop {
            // Check for shutdown signal
            if *self.shutdown_rx.borrow() {
                info!("Worker {} received shutdown signal", self.config.worker_id);
                break;
            }

            match self
                .thingd
                .claim_job(
                    &self.config.queue,
                    &self.config.worker_id,
                    self.config.lease_seconds,
                )
                .await
            {
                Ok(Some(job)) => {
                    info!(
                        "Processing job {} on worker {}",
                        job.id, self.config.worker_id
                    );

                    // Check idempotency (if job has been processed before)
                    if job.attempts > 1 {
                        warn!("Job {} has been attempted {} times", job.id, job.attempts);
                    }

                    match self.handler.handle(job.payload).await {
                        Ok(()) => {
                            if let Err(e) =
                                self.thingd.complete_job(&self.config.queue, &job.id).await
                            {
                                error!("Failed to complete job {}: {}", job.id, e);
                            } else {
                                info!("Job {} completed successfully", job.id);
                            }
                        }
                        Err(e) => {
                            error!("Job {} failed: {}", job.id, e);
                            if let Err(e) = self.thingd.nack_job(&self.config.queue, &job.id).await
                            {
                                error!("Failed to nack job {}: {}", job.id, e);
                            }
                        }
                    }
                }
                Ok(None) => {
                    // No jobs available, wait before polling again
                    tokio::select! {
                        _ = sleep(self.config.poll_interval) => {},
                        _ = self.shutdown_rx.changed() => {
                            if *self.shutdown_rx.borrow() {
                                info!("Worker {} received shutdown signal during sleep", self.config.worker_id);
                                break;
                            }
                        }
                    }
                }
                Err(e) => {
                    error!("Failed to claim job: {}", e);
                    sleep(self.config.poll_interval).await;
                }
            }
        }

        info!("Worker {} shutting down", self.config.worker_id);
    }
}