harn-vm 0.10.52

Async bytecode virtual machine for the Harn programming language
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

use std::collections::BTreeSet;

use serde::{Deserialize, Serialize};

use super::{WorkerQueueClaimHandle, WorkerQueueJob, WorkerQueueResponseRecord};
use crate::triggers::scheduler::{self, SchedulableJob, SchedulerPolicy, SchedulerState};

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkerQueueSummary {
    pub queue: String,
    pub ready: usize,
    pub in_flight: usize,
    pub acked: usize,
    pub purged: usize,
    pub responses: usize,
    pub oldest_unclaimed_age_ms: Option<u64>,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct WorkerQueueJobState {
    pub job_event_id: u64,
    pub enqueued_at_ms: i64,
    pub job: WorkerQueueJob,
    pub active_claim: Option<WorkerQueueClaimHandle>,
    pub acked: bool,
    pub purged: bool,
}

impl WorkerQueueJobState {
    pub fn is_ready(&self) -> bool {
        !self.acked && !self.purged && self.active_claim.is_none()
    }
}

#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
pub struct WorkerQueueState {
    pub queue: String,
    pub responses: Vec<WorkerQueueResponseRecord>,
    pub jobs: Vec<WorkerQueueJobState>,
}

impl WorkerQueueState {
    pub fn summary(&self, now_ms: i64) -> WorkerQueueSummary {
        let ready = self.jobs.iter().filter(|job| job.is_ready()).count();
        let in_flight = self
            .jobs
            .iter()
            .filter(|job| !job.acked && !job.purged && job.active_claim.is_some())
            .count();
        let acked = self.jobs.iter().filter(|job| job.acked).count();
        let purged = self.jobs.iter().filter(|job| job.purged).count();
        let oldest_unclaimed_age_ms = self
            .jobs
            .iter()
            .filter(|job| job.is_ready())
            .map(|job| now_ms.saturating_sub(job.enqueued_at_ms).max(0) as u64)
            .max();
        WorkerQueueSummary {
            queue: self.queue.clone(),
            ready,
            in_flight,
            acked,
            purged,
            responses: self.responses.len(),
            oldest_unclaimed_age_ms,
        }
    }

    /// Select the next ready job under the active policy. Exclusions are
    /// operation-local and never alter durable queue state.
    pub(super) fn next_ready_job_with_scheduler(
        &self,
        scheduler_state: &mut SchedulerState,
        policy: &SchedulerPolicy,
        now_ms: i64,
        excluded_job_event_ids: &BTreeSet<u64>,
    ) -> Option<(&WorkerQueueJobState, scheduler::SchedulerSelection)> {
        let candidates: Vec<&WorkerQueueJobState> = self
            .jobs
            .iter()
            .filter(|job| job.is_ready() && !excluded_job_event_ids.contains(&job.job_event_id))
            .collect();
        if candidates.is_empty() {
            return None;
        }
        let views: Vec<SchedulableJob<'_>> = candidates
            .iter()
            .map(|state| SchedulableJob::from_state(state))
            .collect();

        let in_flight = scheduler::in_flight_by_key(&self.jobs, policy);
        scheduler_state.replace_in_flight(in_flight);

        let pick = scheduler_state.select(&views, policy, now_ms)?;
        candidates
            .into_iter()
            .find(|job| job.job_event_id == pick.job_event_id)
            .map(|job| (job, pick))
    }

    pub(super) fn active_claim_for(&self, job_event_id: u64) -> Option<&WorkerQueueClaimHandle> {
        self.jobs
            .iter()
            .find(|job| job.job_event_id == job_event_id)
            .and_then(|job| job.active_claim.as_ref())
    }
}