ledgence_orchestration_api/
delivery.rs

1use crate::*;
2use ledgence_worker_api::{CloudEvent, ExecutionFailure, ExecutionReport, ProgramDescriptor};
3
4/// UTC milliseconds since the Unix epoch, supplied by the authoritative store.
5pub type Timestamp = u64;
6
7#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
8#[serde(deny_unknown_fields)]
9pub struct Scope {
10    pub tenant_id: String,
11    pub namespace: String,
12}
13impl Scope {
14    pub fn validate(&self) -> Result<()> {
15        validate_text(&self.tenant_id, 128)?;
16        validate_text(&self.namespace, 128)
17    }
18}
19
20#[derive(Debug, Clone, Serialize, Deserialize)]
21#[serde(deny_unknown_fields)]
22pub struct SubmitCommand {
23    pub idempotency_key: String,
24    pub input: SubmitTask,
25    #[serde(skip_serializing_if = "Option::is_none")]
26    pub origin_trace: Option<TraceContext>,
27}
28impl SubmitCommand {
29    /// Decode the portable command at a byte boundary. HTTP adapters may obtain
30    /// its idempotency key from a header, but must use the same strict JSON path.
31    pub fn decode(bytes: &[u8]) -> Result<Self> {
32        let command: Self = decode_unique_json(bytes, SUBMISSION_MAX_BYTES)?;
33        command.input.validate()?;
34        validate_text(&command.idempotency_key, 255)?;
35        Ok(command)
36    }
37}
38
39#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
40#[serde(rename_all = "snake_case")]
41pub enum TaskState {
42    Queued,
43    Active,
44    Succeeded,
45    Failed,
46    Cancelled,
47}
48impl TaskState {
49    pub fn is_terminal(self) -> bool {
50        matches!(self, Self::Succeeded | Self::Failed | Self::Cancelled)
51    }
52}
53#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
54#[serde(rename_all = "snake_case")]
55pub enum AttemptState {
56    Active,
57    Succeeded,
58    Failed,
59    Cancelled,
60    Lost,
61}
62
63/// Transaction-loaded scheduling record. Inputs and the descriptor are immutable.
64#[derive(Debug, Clone, Serialize, Deserialize)]
65pub struct TaskSnapshot {
66    /// Present together only for a nested owned workflow; roots omit both.
67    #[serde(default, skip_serializing_if = "Option::is_none")]
68    pub parent_workflow_id: Option<String>,
69    #[serde(default, skip_serializing_if = "Option::is_none")]
70    pub root_workflow_id: Option<String>,
71
72    /// Workflow execution identity, distinct from the existing per-task run ID.
73    #[serde(default, skip_serializing_if = "Option::is_none")]
74    pub workflow_id: Option<String>,
75    /// Present only for a registered workflow controller task.
76    #[serde(default, skip_serializing_if = "Option::is_none")]
77    pub workflow_activation_id: Option<String>,
78    pub task_id: String,
79    pub run_id: String,
80    pub idempotency_key: String,
81    pub input: SubmitTask,
82    pub descriptor: ProgramDescriptor,
83    pub origin_trace: Option<TraceContext>,
84    pub state: TaskState,
85    pub submitted_at: Timestamp,
86    pub available_at: Timestamp,
87    pub terminal_at: Option<Timestamp>,
88    pub current_attempt_id: Option<String>,
89    pub attempt_count: u32,
90    pub cancel_requested_at: Option<Timestamp>,
91}
92impl TaskSnapshot {
93    pub fn scope(&self) -> Scope {
94        Scope {
95            tenant_id: self.input.tenant_id.clone(),
96            namespace: self.input.namespace.clone(),
97        }
98    }
99}
100
101/// Sessions are issued by the service and never recreated by acquisition.
102/// Unknown/expired session IDs are rejected even after old cursor deletion.
103#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
104pub struct WorkerSession {
105    pub id: String,
106    pub scope: Scope,
107    pub queue: String,
108    pub concurrency: u32,
109    pub expires_at: Timestamp,
110}
111
112#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
113#[serde(deny_unknown_fields)]
114pub struct AcquireCommand {
115    pub scope: Scope,
116    pub queue: String,
117    pub worker_session_id: String,
118    pub consumer_id: u32,
119    pub sequence: u64,
120}
121
122/// Compact replay state: only the latest completed poll per consumer is retained.
123#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
124pub struct ConsumerCursor {
125    pub command: AcquireCommand,
126    pub assignment: Option<AttemptRef>,
127}
128#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
129pub struct AttemptRef {
130    pub task_id: String,
131    pub attempt_id: String,
132}
133
134#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
135#[serde(deny_unknown_fields)]
136pub struct LeaseOwner {
137    pub scope: Scope,
138    pub task_id: String,
139    pub attempt_id: String,
140    pub lease_id: String,
141    pub generation: u32,
142    pub worker_session_id: String,
143    pub consumer_id: u32,
144}
145#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
146pub struct Lease {
147    pub owner: LeaseOwner,
148    pub expires_at: Timestamp,
149}
150
151/// Current authority sampled for this response; never replay a cached TTL.
152#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
153pub struct Authority {
154    pub owner: LeaseOwner,
155    pub expires_at: Timestamp,
156    pub remaining_ms: u64,
157    /// Remaining user-work budget; cleanup/reporting can retain a longer lease.
158    pub execution_remaining_ms: u64,
159    pub renew_sequence: u64,
160    pub cancel_requested: bool,
161    pub dispatch_allowed: bool,
162}
163#[derive(Debug, Clone, Serialize, Deserialize)]
164pub struct Assignment {
165    #[serde(default, skip_serializing_if = "Option::is_none")]
166    pub workflow_activation_id: Option<String>,
167    pub descriptor: ProgramDescriptor,
168    pub event: CloudEvent,
169    pub lease: Lease,
170    pub authority: Authority,
171    pub attempt_deadline: Timestamp,
172}
173impl Assignment {
174    /// The opt-in runtime marker must agree with the immutable invocation event.
175    pub fn validate_workflow_identity(&self) -> Result<()> {
176        let activation = self
177            .event
178            .value()
179            .get("ldgactivationid")
180            .and_then(serde_json::Value::as_str);
181        if self.workflow_activation_id.as_deref() != activation
182            || activation.is_some_and(|id| id != self.lease.owner.task_id)
183        {
184            return Err(ContractError::InvalidInput(
185                "workflow assignment identity mismatch".into(),
186            ));
187        }
188        Ok(())
189    }
190}
191#[derive(Debug, Clone, Serialize, Deserialize)]
192#[serde(tag = "disposition", rename_all = "snake_case")]
193pub enum AcquireReply {
194    Empty {
195        sequence: u64,
196    },
197    Assigned {
198        sequence: u64,
199        assignment: Box<Assignment>,
200    },
201    OwnershipLost {
202        sequence: u64,
203        assignment: AttemptRef,
204    },
205}
206
207#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
208#[serde(rename_all = "snake_case")]
209pub enum RenewIntent {
210    KeepAlive,
211    Dispatch,
212}
213#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
214#[serde(deny_unknown_fields)]
215pub struct RenewCommand {
216    pub owner: LeaseOwner,
217    pub sequence: u64,
218    pub intent: RenewIntent,
219}
220
221/// Required invocation cleanup is done; a healthy warm process may still exist.
222#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
223#[serde(rename_all = "snake_case")]
224pub enum Quiescence {
225    Confirmed,
226    Unconfirmed,
227}
228#[derive(Debug, Clone, Serialize, Deserialize)]
229#[serde(tag = "kind", content = "report", rename_all = "snake_case")]
230pub enum AttemptReport {
231    Completed(ExecutionReport),
232    Failed(ExecutionFailure),
233}
234
235#[derive(Debug, Clone, Serialize, Deserialize)]
236#[serde(deny_unknown_fields)]
237pub struct SettleCommand {
238    pub owner: LeaseOwner,
239    pub operation_id: String,
240    pub report: AttemptReport,
241    pub quiescence: Quiescence,
242    pub processing_trace: Option<TraceContext>,
243}
244impl SettleCommand {
245    /// Preserve JSON numbers and reject duplicate keys before constructing a
246    /// report. The lifecycle core also checks its identity against the attempt.
247    pub fn decode(bytes: &[u8]) -> Result<Self> {
248        let command: Self = decode_unique_json(bytes, SETTLEMENT_MAX_BYTES)?;
249        validate_text(&command.operation_id, 128)?;
250        if let AttemptReport::Completed(report) = &command.report
251            && let ledgence_worker_api::ProgramOutcome::Success { output } = &report.outcome
252        {
253            validate_task_output(output, report.context.identity.activation_id.is_some())?;
254        }
255        Ok(command)
256    }
257}
258#[derive(Debug, Clone, Serialize, Deserialize)]
259pub struct SettlementReceipt {
260    pub operation_id: String,
261    pub task_id: String,
262    pub attempt_id: String,
263    pub accepted_at: Timestamp,
264}
265#[derive(Debug, Clone, Serialize, Deserialize)]
266pub struct AcceptedSettlement {
267    pub command: SettleCommand,
268    pub receipt: SettlementReceipt,
269}
270#[derive(Debug, Clone, Serialize, Deserialize)]
271pub struct SettleReply {
272    pub receipt: SettlementReceipt,
273    pub already_accepted: bool,
274    /// A report can be recorded while required cleanup is still outstanding.
275    pub task_state: TaskState,
276}
277#[derive(Debug, Clone, Serialize, Deserialize)]
278pub struct AttemptSnapshot {
279    pub event: CloudEvent,
280    pub descriptor: ProgramDescriptor,
281    pub lease: Lease,
282    pub deadline: Timestamp,
283    pub authority_deadline: Timestamp,
284    pub state: AttemptState,
285    pub execution_may_have_started: bool,
286    pub last_renewal: Option<RenewCommand>,
287    pub quiescence: Quiescence,
288    pub settlement: Option<AcceptedSettlement>,
289    pub finished_at: Option<Timestamp>,
290}
291
292#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
293#[serde(rename_all = "snake_case")]
294pub enum TransitionReason {
295    Submitted,
296    Claimed,
297    DispatchAuthorized,
298    CancelRequested,
299    Cancelled,
300    ReportAccepted,
301    CleanupConfirmed,
302    Succeeded,
303    Failed,
304    RetryScheduled,
305    LeaseExpired,
306}
307/// Persist with the accompanying records. Per-task ordering is assigned under
308/// the same task lock; heartbeat renewals do not append history rows.
309#[derive(Debug, Clone, Serialize, Deserialize)]
310pub struct HistoryEvent {
311    pub task_id: String,
312    pub attempt_id: Option<String>,
313    pub at: Timestamp,
314    pub reason: TransitionReason,
315}
316#[derive(Debug, Clone, Serialize, Deserialize)]
317pub struct RecordedHistoryEvent {
318    pub sequence: u64,
319    pub event: HistoryEvent,
320}
321
322/// Service boundary implemented by future transport adapters. A successful
323/// mutation reply is permitted only after durable transactional acceptance.
324pub trait TaskService: Send + Sync {
325    /// Claim one exact external dispatch. Successful replies follow durable
326    /// acceptance and bind the complete command. Unsupported implementations
327    /// reject explicitly; they must never fall back to an unrestricted queue scan.
328    fn claim_dispatch<'a>(&'a self, _command: &'a ClaimCommand) -> ContractFuture<'a, ClaimReply> {
329        Box::pin(async {
330            Err(ContractError::InvalidInput(
331                "targeted dispatch claims are unsupported".into(),
332            ))
333        })
334    }
335
336    fn open_session<'a>(
337        &'a self,
338        scope: &'a Scope,
339        queue: &'a str,
340        concurrency: u32,
341    ) -> ContractFuture<'a, WorkerSession>;
342    fn extend_session<'a>(
343        &'a self,
344        worker_session_id: &'a str,
345    ) -> ContractFuture<'a, WorkerSession>;
346    fn submit<'a>(&'a self, command: &'a SubmitCommand) -> ContractFuture<'a, TaskSnapshot>;
347    /// Read one bounded page of matching committed task statuses in descending
348    /// submission-time/task-ID order. Each page has its own read snapshot.
349    fn list_tasks<'a>(
350        &'a self,
351        scope: &'a Scope,
352        query: &'a TaskListQuery,
353    ) -> ContractFuture<'a, TaskPage>;
354    /// Read compact scheduling metadata without application payloads.
355    fn status<'a>(&'a self, scope: &'a Scope, task_id: &'a str) -> ContractFuture<'a, TaskStatus>;
356    /// Read task metadata and its logical outcome from one consistent snapshot.
357    fn result<'a>(&'a self, scope: &'a Scope, task_id: &'a str) -> ContractFuture<'a, TaskResult>;
358    fn inspect<'a>(
359        &'a self,
360        scope: &'a Scope,
361        task_id: &'a str,
362    ) -> ContractFuture<'a, TaskSnapshot>;
363    fn inspect_attempt<'a>(
364        &'a self,
365        scope: &'a Scope,
366        task_id: &'a str,
367        attempt_id: &'a str,
368    ) -> ContractFuture<'a, AttemptSnapshot>;
369    /// Return at most 100 ordered records after the supplied sequence. Large
370    /// application outcomes are fetched through result or inspect_attempt, not history.
371    fn history<'a>(
372        &'a self,
373        scope: &'a Scope,
374        task_id: &'a str,
375        after_sequence: u64,
376    ) -> ContractFuture<'a, Vec<RecordedHistoryEvent>>;
377    fn acquire<'a>(
378        &'a self,
379        command: &'a AcquireCommand,
380        options: AcquireOptions,
381    ) -> ContractFuture<'a, AcquireReply>;
382    fn renew<'a>(&'a self, command: &'a RenewCommand) -> ContractFuture<'a, Authority>;
383    fn settle<'a>(&'a self, command: &'a SettleCommand) -> ContractFuture<'a, SettleReply>;
384    fn confirm_quiescence<'a>(&'a self, owner: &'a LeaseOwner) -> ContractFuture<'a, TaskState>;
385    fn cancel<'a>(&'a self, scope: &'a Scope, task_id: &'a str) -> ContractFuture<'a, TaskState>;
386}
ledgence_orchestration_api/delivery.rs

ledgence_orchestration_api/
delivery.rs
