ledgence_orchestration_api/
completion.rs

1//! One-shot durable completion subscriptions. Transport I/O is outside storage transactions.
2use crate::*;
3use serde_json::Value;
4use std::time::Instant;
5
6pub const COMPLETION_EVENT_MAX_BYTES: usize = 16 * 1024;
7pub const COMPLETION_COMMAND_MAX_BYTES: usize = 4096;
8pub const COMPLETION_STATUS_MAX_BYTES: usize = 24 * 1024;
9pub const MAX_COMPLETION_SUBSCRIPTIONS: u32 = 16;
10pub const MAX_COMPLETION_BATCH: u32 = 16;
11pub const COMPLETION_MAX_ATTEMPTS: u32 = 8;
12pub const COMPLETION_MAX_GENERATION: u32 = 1000;
13pub const COMPLETION_LEASE_MS: u64 = 30_000;
14pub const COMPLETION_MAX_RETRY_DELAY_MS: u64 = 300_000;
15const MAX_TIME: u64 = 253_402_300_799_999;
16
17#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
18#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
19pub enum CompletionTarget {
20    Task { id: String },
21    Workflow { id: String },
22}
23impl CompletionTarget {
24    pub fn id(&self) -> &str {
25        match self {
26            Self::Task { id } | Self::Workflow { id } => id,
27        }
28    }
29    pub fn kind(&self) -> &'static str {
30        match self {
31            Self::Task { .. } => "task",
32            Self::Workflow { .. } => "workflow",
33        }
34    }
35    pub fn validate(&self) -> Result<()> {
36        validate_text(self.id(), 128)
37    }
38}
39
40#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
41#[serde(deny_unknown_fields)]
42pub struct CompletionSubscribeCommand {
43    pub scope: Scope,
44    pub target: CompletionTarget,
45    pub destination: String,
46    /// Unique within scope + target, not across unrelated executions.
47    pub idempotency_key: String,
48}
49impl CompletionSubscribeCommand {
50    pub fn validate(&self) -> Result<()> {
51        self.scope.validate()?;
52        self.target.validate()?;
53        validate_text(&self.destination, 128)?;
54        validate_text(&self.idempotency_key, 128)?;
55        bounded(self, COMPLETION_COMMAND_MAX_BYTES)
56    }
57    pub fn decode(bytes: &[u8]) -> Result<Self> {
58        let command: Self = decode_unique_json(bytes, COMPLETION_COMMAND_MAX_BYTES)?;
59        command.validate()?;
60        Ok(command)
61    }
62}
63
64#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
65#[serde(deny_unknown_fields)]
66pub struct CompletionRetryCommand {
67    pub scope: Scope,
68    pub subscription_id: String,
69    /// The exhausted generation to rearm. This is the durable operation identity.
70    /// Replaying an older accepted generation returns current state without rearming.
71    pub expected_generation: u32,
72}
73impl CompletionRetryCommand {
74    pub fn validate(&self) -> Result<()> {
75        self.scope.validate()?;
76        validate_text(&self.subscription_id, 128)?;
77        if !(1..COMPLETION_MAX_GENERATION).contains(&self.expected_generation) {
78            return Err(invalid(
79                "completion retry generation is outside supported bounds",
80            ));
81        }
82        Ok(())
83    }
84    pub fn decode(bytes: &[u8]) -> Result<Self> {
85        let command: Self = decode_unique_json(bytes, COMPLETION_COMMAND_MAX_BYTES)?;
86        command.validate()?;
87        Ok(command)
88    }
89}
90
91/// Operator-configured immutable binding. Core does not interpret transport settings.
92#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
93#[serde(deny_unknown_fields)]
94pub struct CompletionDestination {
95    pub scope: Scope,
96    pub destination: String,
97    pub binding: String,
98}
99impl CompletionDestination {
100    pub fn validate(&self) -> Result<()> {
101        self.scope.validate()?;
102        validate_text(&self.destination, 128)?;
103        validate_text(&self.binding, 4096)
104    }
105}
106
107/// Compact reference-only CloudEvent: no application result copy and no platform `data`.
108/// Event identity and bytes remain unchanged across retries and manual redelivery.
109#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
110#[serde(transparent)]
111pub struct CompletionEvent(Value);
112impl CompletionEvent {
113    pub fn new(value: Value) -> Result<Self> {
114        let event = Self(value);
115        event.validate()?;
116        Ok(event)
117    }
118    pub fn value(&self) -> &Value {
119        &self.0
120    }
121    pub fn id(&self) -> &str {
122        self.0["id"].as_str().unwrap_or_default()
123    }
124    pub fn source(&self) -> &str {
125        self.0["source"].as_str().unwrap_or_default()
126    }
127    pub fn trace_context(&self) -> Option<TraceContext> {
128        self.0
129            .get("traceparent")
130            .and_then(Value::as_str)
131            .map(|traceparent| TraceContext {
132                traceparent: traceparent.to_owned(),
133                tracestate: self
134                    .0
135                    .get("tracestate")
136                    .and_then(Value::as_str)
137                    .map(str::to_owned),
138            })
139    }
140    pub fn decode(bytes: &[u8]) -> Result<Self> {
141        Self::new(decode_unique_json(bytes, COMPLETION_EVENT_MAX_BYTES)?)
142    }
143    pub fn validate(&self) -> Result<()> {
144        bounded(&self.0, COMPLETION_EVENT_MAX_BYTES)?;
145        ledgence_worker_api::validate_cloudevent_context(&self.0)?;
146        let object = self
147            .0
148            .as_object()
149            .ok_or_else(|| invalid("completion event must be an object"))?;
150        const ALLOWED: &[&str] = &[
151            "specversion",
152            "id",
153            "source",
154            "type",
155            "subject",
156            "time",
157            "ldgtenantid",
158            "ldgnamespace",
159            "ldgstate",
160            "ldgresultref",
161            "ldgtaskid",
162            "ldgrunid",
163            "ldgattemptid",
164            "ldgworkflowid",
165            "ldgactivationid",
166            "ldgparentworkflowid",
167            "ldgrootworkflowid",
168            "ldgcorrelationkey",
169            "ldgcorrelationkeyencoding",
170            "traceparent",
171            "tracestate",
172        ];
173        if object.keys().any(|key| !ALLOWED.contains(&key.as_str())) {
174            return Err(invalid("unsupported completion event attribute"));
175        }
176        for name in [
177            "id",
178            "subject",
179            "time",
180            "ldgtenantid",
181            "ldgnamespace",
182            "ldgstate",
183            "ldgresultref",
184        ] {
185            validate_text(
186                text(&self.0, name)?,
187                if name == "ldgresultref" { 2048 } else { 256 },
188            )?;
189        }
190        if self.source() != "urn:ledgence:orchestrator"
191            || !matches!(
192                text(&self.0, "ldgstate")?,
193                "succeeded" | "failed" | "cancelled"
194            )
195        {
196            return Err(invalid("invalid completion source or terminal state"));
197        }
198        let target = match text(&self.0, "type")? {
199            "com.ledgence.task.completed.v1" => {
200                validate_text(text(&self.0, "ldgrunid")?, 128)?;
201                CompletionTarget::Task {
202                    id: text(&self.0, "ldgtaskid")?.to_owned(),
203                }
204            }
205            "com.ledgence.workflow.completed.v1" => {
206                if ["ldgtaskid", "ldgrunid", "ldgattemptid", "ldgactivationid"]
207                    .iter()
208                    .any(|key| object.contains_key(*key))
209                {
210                    return Err(invalid(
211                        "workflow completion must not impersonate a controller task",
212                    ));
213                }
214                CompletionTarget::Workflow {
215                    id: text(&self.0, "ldgworkflowid")?.to_owned(),
216                }
217            }
218            _ => return Err(invalid("invalid completion event type")),
219        };
220        target.validate()?;
221        let scope = Scope {
222            tenant_id: text(&self.0, "ldgtenantid")?.into(),
223            namespace: text(&self.0, "ldgnamespace")?.into(),
224        };
225        scope.validate()?;
226        if self.id() != completion_event_id(&target)
227            || text(&self.0, "subject")? != format!("{}s/{}", target.kind(), target.id())
228            || text(&self.0, "ldgresultref")? != completion_result_ref(&scope, &target)
229        {
230            return Err(invalid("completion event reference identity differs"));
231        }
232        for key in [
233            "ldgtaskid",
234            "ldgrunid",
235            "ldgattemptid",
236            "ldgworkflowid",
237            "ldgactivationid",
238            "ldgparentworkflowid",
239            "ldgrootworkflowid",
240        ] {
241            if let Some(value) = object.get(key) {
242                validate_text(
243                    value
244                        .as_str()
245                        .ok_or_else(|| invalid("invalid completion identifier"))?,
246                    128,
247                )?;
248            }
249        }
250        validate_workflow_lineage(
251            object.get("ldgworkflowid").and_then(Value::as_str),
252            object.get("ldgparentworkflowid").and_then(Value::as_str),
253            object.get("ldgrootworkflowid").and_then(Value::as_str),
254        )?;
255        if (object.contains_key("ldgactivationid")
256            && (!object.contains_key("ldgworkflowid")
257                || self.0["ldgactivationid"] != self.0["ldgtaskid"]))
258            || (self.0["ldgstate"] == "cancelled" && object.contains_key("ldgattemptid"))
259        {
260            return Err(invalid("invalid completion lineage or deciding attempt"));
261        }
262        if let Some(key) = object.get("ldgcorrelationkey") {
263            let key = key
264                .as_str()
265                .ok_or_else(|| invalid("invalid completion correlation key"))?;
266            match object.get("ldgcorrelationkeyencoding") {
267                None if key.len() <= 512 && !key.chars().any(char::is_control) => {}
268                Some(Value::String(encoding)) if encoding == "percent" => {
269                    decode_completion_correlation(key)?;
270                }
271                _ => return Err(invalid("invalid completion correlation encoding or length")),
272            }
273        } else if object.contains_key("ldgcorrelationkeyencoding") {
274            return Err(invalid("correlation encoding requires a key"));
275        }
276        Ok(())
277    }
278    pub fn matches(&self, scope: &Scope, target: &CompletionTarget) -> bool {
279        self.0["ldgtenantid"] == scope.tenant_id
280            && self.0["ldgnamespace"] == scope.namespace
281            && self.id() == completion_event_id(target)
282            && self.0["type"] == format!("com.ledgence.{}.completed.v1", target.kind())
283    }
284}
285
286pub fn completion_event_id(target: &CompletionTarget) -> String {
287    format!("evt_{}_completed_{}", target.kind(), target.id())
288}
289pub fn completion_result_ref(scope: &Scope, target: &CompletionTarget) -> String {
290    format!(
291        "/v1/{}s/result?tenant_id={}&namespace={}&{}_id={}",
292        target.kind(),
293        percent(&scope.tenant_id),
294        percent(&scope.namespace),
295        target.kind(),
296        percent(target.id())
297    )
298}
299fn percent(value: &str) -> String {
300    let mut output = String::new();
301    const HEX: &[u8] = b"0123456789ABCDEF";
302    for byte in value.bytes() {
303        if byte.is_ascii_alphanumeric() || b"-._~".contains(&byte) {
304            output.push(char::from(byte));
305        } else {
306            output.push('%');
307            output.push(char::from(HEX[(byte >> 4) as usize]));
308            output.push(char::from(HEX[(byte & 15) as usize]));
309        }
310    }
311    output
312}
313
314/// Encode the rare accepted business strings excluded by CloudEvents metadata.
315pub fn encode_completion_correlation(value: &str) -> String {
316    percent(value)
317}
318pub fn decode_completion_correlation(value: &str) -> Result<String> {
319    if value.len() > 1536 {
320        return Err(invalid("encoded completion correlation exceeds limit"));
321    }
322    let mut bytes = Vec::with_capacity(value.len());
323    let mut iter = value.bytes();
324    while let Some(byte) = iter.next() {
325        if byte == b'%' {
326            let high = iter.next().and_then(|c| char::from(c).to_digit(16));
327            let low = iter.next().and_then(|c| char::from(c).to_digit(16));
328            match (high, low) {
329                (Some(high), Some(low)) => bytes.push((high * 16 + low) as u8),
330                _ => return Err(invalid("malformed completion correlation escape")),
331            }
332        } else {
333            bytes.push(byte);
334        }
335    }
336    let decoded = String::from_utf8(bytes).map_err(|_| invalid("correlation is not UTF-8"))?;
337    if decoded.len() > 512 || decoded.chars().any(char::is_control) || percent(&decoded) != value {
338        return Err(invalid(
339            "invalid or noncanonical encoded completion correlation",
340        ));
341    }
342    Ok(decoded)
343}
344
345#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
346#[serde(rename_all = "snake_case")]
347pub enum CompletionState {
348    Waiting,
349    Pending,
350    Delivering,
351    Retrying,
352    Delivered,
353    Exhausted,
354}
355
356#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
357#[serde(deny_unknown_fields)]
358pub struct CompletionSubscription {
359    pub subscription_id: String,
360    pub command: CompletionSubscribeCommand,
361    pub state: CompletionState,
362    pub generation: u32,
363    /// Leased transport attempts in this generation, including uncertain sends.
364    pub attempts: u32,
365    pub total_attempts: u64,
366    pub created_at: Timestamp,
367    #[serde(deserialize_with = "crate::observation::required_option")]
368    pub activated_at: Option<Timestamp>,
369    #[serde(deserialize_with = "crate::observation::required_option")]
370    pub next_attempt_at: Option<Timestamp>,
371    #[serde(deserialize_with = "crate::observation::required_option")]
372    pub lease_expires_at: Option<Timestamp>,
373    #[serde(deserialize_with = "crate::observation::required_option")]
374    pub delivered_at: Option<Timestamp>,
375    #[serde(deserialize_with = "crate::observation::required_option")]
376    pub exhausted_at: Option<Timestamp>,
377    #[serde(deserialize_with = "crate::observation::required_option")]
378    pub last_failure: Option<String>,
379    #[serde(deserialize_with = "crate::observation::required_option")]
380    pub event: Option<CompletionEvent>,
381}
382impl CompletionSubscription {
383    pub fn validate(&self) -> Result<()> {
384        self.command.validate()?;
385        validate_text(&self.subscription_id, 128)?;
386        if !(1..=COMPLETION_MAX_GENERATION).contains(&self.generation)
387            || self.attempts > COMPLETION_MAX_ATTEMPTS
388            || self.total_attempts < u64::from(self.attempts)
389            || self.total_attempts > u64::from(self.generation * COMPLETION_MAX_ATTEMPTS)
390            || self.created_at > MAX_TIME
391        {
392            return Err(invalid("invalid completion subscription counters or time"));
393        }
394        for time in [
395            self.activated_at,
396            self.next_attempt_at,
397            self.lease_expires_at,
398            self.delivered_at,
399            self.exhausted_at,
400        ]
401        .into_iter()
402        .flatten()
403        {
404            if time < self.created_at || time > MAX_TIME {
405                return Err(invalid("invalid completion subscription timestamp"));
406            }
407        }
408        if let Some(reason) = &self.last_failure {
409            validate_text(reason, 256)?;
410        }
411        let active = self.event.is_some() && self.activated_at.is_some();
412        let no_terminal = self.delivered_at.is_none() && self.exhausted_at.is_none();
413        let valid = match self.state {
414            CompletionState::Waiting => {
415                !active
416                    && self.event.is_none()
417                    && self.activated_at.is_none()
418                    && self.next_attempt_at.is_none()
419                    && self.lease_expires_at.is_none()
420                    && no_terminal
421                    && self.attempts == 0
422                    && self.total_attempts == 0
423                    && self.generation == 1
424                    && self.last_failure.is_none()
425            }
426            CompletionState::Pending => {
427                active
428                    && self.attempts == 0
429                    && self.next_attempt_at.is_some()
430                    && self.lease_expires_at.is_none()
431                    && no_terminal
432            }
433            CompletionState::Retrying => {
434                active
435                    && (1..COMPLETION_MAX_ATTEMPTS).contains(&self.attempts)
436                    && self.next_attempt_at.is_some()
437                    && self.lease_expires_at.is_none()
438                    && no_terminal
439            }
440            CompletionState::Delivering => {
441                active
442                    && self.attempts > 0
443                    && self.next_attempt_at.is_none()
444                    && self.lease_expires_at.is_some()
445                    && no_terminal
446            }
447            CompletionState::Delivered => {
448                active
449                    && self.attempts > 0
450                    && self.next_attempt_at.is_none()
451                    && self.lease_expires_at.is_none()
452                    && self.delivered_at.is_some()
453                    && self.exhausted_at.is_none()
454            }
455            CompletionState::Exhausted => {
456                active
457                    && self.attempts == COMPLETION_MAX_ATTEMPTS
458                    && self.next_attempt_at.is_none()
459                    && self.lease_expires_at.is_none()
460                    && self.delivered_at.is_none()
461                    && self.exhausted_at.is_some()
462            }
463        };
464        if !valid {
465            return Err(invalid("inconsistent completion delivery state"));
466        }
467        if let Some(event) = &self.event {
468            event.validate()?;
469            if !event.matches(&self.command.scope, &self.command.target) {
470                return Err(invalid("completion event target mismatch"));
471            }
472        }
473        bounded(self, COMPLETION_STATUS_MAX_BYTES)
474    }
475    pub fn matches(&self, command: &CompletionSubscribeCommand) -> bool {
476        self.command == *command
477    }
478}
479
480#[derive(Debug, Clone)]
481pub struct CompletionLease {
482    pub subscription: CompletionSubscription,
483    pub lease_token: String,
484    /// Exact committed CloudEvent bytes, reused without regeneration on every send.
485    pub event_bytes: Vec<u8>,
486}
487impl CompletionLease {
488    pub fn validate(&self) -> Result<()> {
489        self.subscription.validate()?;
490        validate_text(&self.lease_token, 128)?;
491        if self.subscription.state != CompletionState::Delivering
492            || self.subscription.event.as_ref()
493                != Some(&CompletionEvent::decode(&self.event_bytes)?)
494        {
495            return Err(invalid("invalid completion lease event or state"));
496        }
497        Ok(())
498    }
499}
500#[derive(Debug, Clone, PartialEq, Eq)]
501pub enum CompletionDeliveryOutcome {
502    Confirmed,
503    Retry {
504        reason: String,
505        retry_after_ms: Option<u64>,
506    },
507}
508#[derive(Debug, Clone)]
509pub struct CompletionDeliveryResult {
510    pub subscription_id: String,
511    pub generation: u32,
512    pub lease_token: String,
513    pub outcome: CompletionDeliveryOutcome,
514}
515impl CompletionDeliveryResult {
516    pub fn validate(&self) -> Result<()> {
517        validate_text(&self.subscription_id, 128)?;
518        validate_text(&self.lease_token, 128)?;
519        if !(1..=COMPLETION_MAX_GENERATION).contains(&self.generation) {
520            return Err(invalid("invalid completion generation"));
521        }
522        if let CompletionDeliveryOutcome::Retry {
523            reason,
524            retry_after_ms,
525        } = &self.outcome
526        {
527            validate_text(reason, 256)?;
528            if retry_after_ms.is_some_and(|delay| delay > COMPLETION_MAX_RETRY_DELAY_MS) {
529                return Err(invalid("completion retry delay exceeds bound"));
530            }
531        }
532        Ok(())
533    }
534}
535
536/// Subscriber-facing operations. Accepted subscriptions survive caller disconnection.
537pub trait CompletionService: Send + Sync {
538    fn subscribe_completion<'a>(
539        &'a self,
540        command: &'a CompletionSubscribeCommand,
541    ) -> ContractFuture<'a, CompletionSubscription>;
542    fn completion_status<'a>(
543        &'a self,
544        scope: &'a Scope,
545        subscription_id: &'a str,
546    ) -> ContractFuture<'a, CompletionSubscription>;
547    fn retry_completion<'a>(
548        &'a self,
549        command: &'a CompletionRetryCommand,
550    ) -> ContractFuture<'a, CompletionSubscription>;
551}
552/// Atomic registration and terminal hooks share the execution row lock. Delivery
553/// leasing only locks subscription rows; it must never lock executions afterward.
554pub trait CompletionStore: Send + Sync {
555    fn configure_completion_destination<'a>(
556        &'a self,
557        destination: &'a CompletionDestination,
558    ) -> ContractFuture<'a, ()>;
559    fn subscribe_completion<'a>(
560        &'a self,
561        command: &'a CompletionSubscribeCommand,
562    ) -> ContractFuture<'a, CompletionSubscription>;
563    fn completion_status<'a>(
564        &'a self,
565        scope: &'a Scope,
566        subscription_id: &'a str,
567    ) -> ContractFuture<'a, CompletionSubscription>;
568    fn retry_completion<'a>(
569        &'a self,
570        command: &'a CompletionRetryCommand,
571    ) -> ContractFuture<'a, CompletionSubscription>;
572    fn lease_completions<'a>(
573        &'a self,
574        destination: &'a CompletionDestination,
575        limit: u32,
576        deadline: Instant,
577    ) -> ContractFuture<'a, Vec<CompletionLease>>;
578    fn complete_deliveries<'a>(
579        &'a self,
580        completions: &'a [CompletionDeliveryResult],
581        deadline: Instant,
582    ) -> ContractFuture<'a, ()>;
583}
584pub trait CompletionSender: Send + Sync {
585    /// Positive transport acknowledgment only; ambiguous outcomes require retry.
586    fn deliver<'a>(
587        &'a self,
588        lease: &'a CompletionLease,
589        deadline: Instant,
590    ) -> ContractFuture<'a, CompletionDeliveryOutcome>;
591}
592fn text<'a>(value: &'a Value, key: &str) -> Result<&'a str> {
593    value
594        .get(key)
595        .and_then(Value::as_str)
596        .ok_or_else(|| invalid("missing completion event text attribute"))
597}
598fn invalid(message: &str) -> ContractError {
599    ContractError::InvalidInput(message.into())
600}
601fn bounded(value: &impl Serialize, maximum: usize) -> Result<()> {
602    crate::submission::check_encoded_size(value, maximum, "completion contract").map_err(Into::into)
603}
604
605#[cfg(test)]
606mod tests;
ledgence_orchestration_api/completion.rs

ledgence_orchestration_api/
completion.rs
