ledgence_orchestration_api/
acquisition.rs

1//! Local acquisition budgets and optional advisory wake signals.
2
3use crate::*;
4use std::time::{Duration, Instant};
5
6/// A wait preference and the enclosing local exchange deadline. Neither is part
7/// of durable acquisition identity; only the nonzero wait preference is sent
8/// over a transport. Every caller must keep its original deadline across probes.
9#[derive(Debug, Clone, Copy)]
10pub struct AcquireOptions {
11    pub max_wait: Duration,
12    pub deadline: Instant,
13}
14impl AcquireOptions {
15    pub fn new(max_wait: Duration, deadline: Instant) -> Result<Self> {
16        let options = Self { max_wait, deadline };
17        options.validate()?;
18        Ok(options)
19    }
20
21    pub fn immediate(deadline: Instant) -> Self {
22        Self {
23            max_wait: Duration::ZERO,
24            deadline,
25        }
26    }
27
28    pub fn for_wait(max_wait: Duration) -> Result<Self> {
29        Self::new(
30            max_wait,
31            Instant::now() + Duration::from_millis(CONTROL_REQUEST_TIMEOUT_MS),
32        )
33    }
34
35    pub fn validate(&self) -> Result<()> {
36        if self.max_wait > Duration::from_millis(LONG_POLL_WAIT_MS) {
37            return Err(ContractError::InvalidInput(
38                "acquisition wait exceeds 20000 ms".into(),
39            ));
40        }
41        Ok(())
42    }
43}
44
45/// Internal provenance prevents replayed assignments from implying more backlog.
46#[derive(Debug, Clone, Copy, PartialEq, Eq)]
47pub enum AcquisitionCompletion {
48    Replayed,
49    Claimed,
50    FinalizedEmpty,
51}
52
53#[derive(Debug, Clone)]
54pub enum AcquisitionProbe {
55    Completed {
56        reply: AcquireReply,
57        kind: AcquisitionCompletion,
58    },
59    Pending {
60        session_remaining_ms: u64,
61    },
62}
63
64#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
65#[serde(deny_unknown_fields)]
66pub struct AcquisitionQueue {
67    pub scope: Scope,
68    pub queue: String,
69}
70impl From<&AcquireCommand> for AcquisitionQueue {
71    fn from(command: &AcquireCommand) -> Self {
72        Self {
73            scope: command.scope.clone(),
74            queue: command.queue.clone(),
75        }
76    }
77}
78
79#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
80#[serde(deny_unknown_fields)]
81pub struct AcquisitionKey {
82    pub queue: AcquisitionQueue,
83    pub worker_session_id: String,
84    pub consumer_id: u32,
85    pub sequence: u64,
86}
87impl From<&AcquireCommand> for AcquisitionKey {
88    fn from(command: &AcquireCommand) -> Self {
89        Self {
90            queue: command.into(),
91            worker_session_id: command.worker_session_id.clone(),
92            consumer_id: command.consumer_id,
93            sequence: command.sequence,
94        }
95    }
96}
97
98/// Hints carry identities only. They never grant execution authority or cache a
99/// reply. A rescan follows notification subscription/reconnection.
100#[derive(Debug, Clone, PartialEq, Eq, Hash)]
101pub enum AcquisitionHint {
102    QueueChanged(AcquisitionQueue),
103    AcquisitionCompleted(AcquisitionKey),
104    Rescan,
105}
106
107/// Optional adapter-to-service wake port. Implementations must return promptly
108/// without network I/O and bound retained interests. Periodic fallback remains
109/// necessary even when a transport delivers these hints.
110pub trait AcquisitionWake: Send + Sync {
111    fn wake(&self, hint: AcquisitionHint);
112}
ledgence_orchestration_api/acquisition.rs

ledgence_orchestration_api/
acquisition.rs
