cloacina 0.8.0

A Rust library for resilient task execution and orchestration.
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
108
109
110
111
112
113
114
115
116
117
118
119
120
121

/*
 *  Copyright 2025-2026 Colliery Software
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

//! Delivery Outbox Model
//!
//! Domain types for the delivery outbox — the durable, ack-tracked,
//! recipient-addressed push-delivery outbox of the interservice communication
//! substrate (spec CLOACI-S-0012, decided in CLOACI-A-0006).
//!
//! Unlike [`crate::models::task_outbox`] (a transient, competing-consumer
//! claim queue deleted on claim), delivery-outbox rows are addressed to a
//! specific recipient, carry a payload, and are retained until acked. The
//! substrate is Postgres-only at runtime.

use crate::database::universal_types::UniversalTimestamp;
use serde::{Deserialize, Serialize};

/// Delivery lifecycle of an outbox row.
///
/// `Pending` → `Delivered` (pushed to recipient, awaiting ack) → `Acked`
/// (recipient confirmed receipt). `Delivered` → `Pending` is the sweeper /
/// reconnect redelivery path (T-0628 / T-0627).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum DeliveryState {
    Pending,
    Delivered,
    Acked,
}

impl DeliveryState {
    /// String form persisted in the `delivery_state` column.
    pub fn as_str(&self) -> &'static str {
        match self {
            DeliveryState::Pending => "pending",
            DeliveryState::Delivered => "delivered",
            DeliveryState::Acked => "acked",
        }
    }

    /// Parse from the persisted string form.
    pub fn from_db(s: &str) -> Option<DeliveryState> {
        match s {
            "pending" => Some(DeliveryState::Pending),
            "delivered" => Some(DeliveryState::Delivered),
            "acked" => Some(DeliveryState::Acked),
            _ => None,
        }
    }

    /// Whether `self → next` is a permitted transition.
    ///
    /// - `Pending → Delivered`: relay pushed the row.
    /// - `Delivered → Acked`: recipient confirmed.
    /// - `Delivered → Pending`: redelivery (sweeper reclaim / reconnect resync).
    ///
    /// `Acked` is terminal. All other transitions are rejected.
    pub fn can_transition_to(self, next: DeliveryState) -> bool {
        use DeliveryState::*;
        matches!(
            (self, next),
            (Pending, Delivered) | (Delivered, Acked) | (Delivered, Pending)
        )
    }
}

/// A delivery-outbox row (domain type).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DeliveryOutbox {
    /// Auto-incrementing primary key (also the replay-ordering cursor).
    pub id: i64,
    /// Addressed recipient key (e.g. `agent:<uuid>`, `exec_events:<exec_id>`).
    pub recipient: String,
    /// Payload discriminator (e.g. `work`, `execution_event`).
    pub kind: String,
    /// Tenant scope, when applicable (matches the server's `Nullable<Text>` tenant id).
    pub tenant_id: Option<String>,
    /// Opaque payload bytes. NOTIFY never carries this — only the row id.
    pub payload: Vec<u8>,
    /// Current delivery lifecycle state (see [`DeliveryState`]).
    pub delivery_state: String,
    /// Number of delivery attempts so far (incremented on each (re)delivery).
    pub delivery_attempts: i32,
    /// When the row was enqueued.
    pub created_at: UniversalTimestamp,
    /// When it was last pushed to the recipient.
    pub delivered_at: Option<UniversalTimestamp>,
    /// When the recipient acked.
    pub acked_at: Option<UniversalTimestamp>,
}

impl DeliveryOutbox {
    /// Typed view of the persisted `delivery_state` string.
    pub fn state(&self) -> Option<DeliveryState> {
        DeliveryState::from_db(&self.delivery_state)
    }
}

/// Structure for enqueuing a new delivery-outbox row.
///
/// `delivery_state` defaults to `pending` and counters/timestamps are set by
/// the DAL; callers supply only the addressing and payload.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NewDeliveryOutbox {
    pub recipient: String,
    pub kind: String,
    pub tenant_id: Option<String>,
    pub payload: Vec<u8>,
}