harn_vm/triggers/worker_queue/
state.rs

1use std::collections::BTreeSet;
2
3use serde::{Deserialize, Serialize};
4
5use super::{TenantClaimScope, WorkerQueueClaimHandle, WorkerQueueJob, WorkerQueueResponseRecord};
6use crate::triggers::scheduler::{self, SchedulableJob, SchedulerPolicy, SchedulerState};
7
8#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
9pub struct WorkerQueueSummary {
10    pub queue: String,
11    pub ready: usize,
12    pub in_flight: usize,
13    pub acked: usize,
14    pub purged: usize,
15    pub responses: usize,
16    pub oldest_unclaimed_age_ms: Option<u64>,
17}
18
19#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
20pub struct WorkerQueueJobState {
21    pub job_event_id: u64,
22    pub enqueued_at_ms: i64,
23    pub job: WorkerQueueJob,
24    pub active_claim: Option<WorkerQueueClaimHandle>,
25    pub acked: bool,
26    pub purged: bool,
27}
28
29impl WorkerQueueJobState {
30    pub fn is_ready(&self) -> bool {
31        !self.acked && !self.purged && self.active_claim.is_none()
32    }
33}
34
35#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
36pub struct WorkerQueueState {
37    pub queue: String,
38    pub responses: Vec<WorkerQueueResponseRecord>,
39    pub jobs: Vec<WorkerQueueJobState>,
40}
41
42impl WorkerQueueState {
43    pub fn summary(&self, now_ms: i64) -> WorkerQueueSummary {
44        let ready = self.jobs.iter().filter(|job| job.is_ready()).count();
45        let in_flight = self
46            .jobs
47            .iter()
48            .filter(|job| !job.acked && !job.purged && job.active_claim.is_some())
49            .count();
50        let acked = self.jobs.iter().filter(|job| job.acked).count();
51        let purged = self.jobs.iter().filter(|job| job.purged).count();
52        let oldest_unclaimed_age_ms = self
53            .jobs
54            .iter()
55            .filter(|job| job.is_ready())
56            .map(|job| now_ms.saturating_sub(job.enqueued_at_ms).max(0) as u64)
57            .max();
58        WorkerQueueSummary {
59            queue: self.queue.clone(),
60            ready,
61            in_flight,
62            acked,
63            purged,
64            responses: self.responses.len(),
65            oldest_unclaimed_age_ms,
66        }
67    }
68
69    /// Select the next ready job under the active policy. Exclusions are
70    /// operation-local and never alter durable queue state.
71    pub(super) fn next_ready_job_with_scheduler(
72        &self,
73        scheduler_state: &mut SchedulerState,
74        policy: &SchedulerPolicy,
75        now_ms: i64,
76        excluded_job_event_ids: &BTreeSet<u64>,
77        tenant_scope: TenantClaimScope<'_>,
78    ) -> Option<(&WorkerQueueJobState, scheduler::SchedulerSelection)> {
79        let candidates: Vec<&WorkerQueueJobState> = self
80            .jobs
81            .iter()
82            .filter(|job| {
83                job.is_ready()
84                    && !excluded_job_event_ids.contains(&job.job_event_id)
85                    && match tenant_scope {
86                        TenantClaimScope::Any => true,
87                        TenantClaimScope::Untenanted => job.job.event.tenant_id.is_none(),
88                        TenantClaimScope::Tenant(tenant_id) => {
89                            job.job.event.tenant_id.as_ref() == Some(tenant_id)
90                        }
91                    }
92            })
93            .collect();
94        if candidates.is_empty() {
95            return None;
96        }
97        let views: Vec<SchedulableJob<'_>> = candidates
98            .iter()
99            .map(|state| SchedulableJob::from_state(state))
100            .collect();
101
102        let in_flight = scheduler::in_flight_by_key(&self.jobs, policy);
103        scheduler_state.replace_in_flight(in_flight);
104
105        let pick = scheduler_state.select(&views, policy, now_ms)?;
106        candidates
107            .into_iter()
108            .find(|job| job.job_event_id == pick.job_event_id)
109            .map(|job| (job, pick))
110    }
111
112    pub(super) fn active_claim_for(&self, job_event_id: u64) -> Option<&WorkerQueueClaimHandle> {
113        self.jobs
114            .iter()
115            .find(|job| job.job_event_id == job_event_id)
116            .and_then(|job| job.active_claim.as_ref())
117    }
118}
harn_vm/triggers/worker_queue/state.rs

harn_vm/triggers/worker_queue/
state.rs
