cortexkit_bus_trait/
work_queue.rs

1use std::time::Duration;
2
3use async_trait::async_trait;
4
5use crate::{BusResult, ContentDigest, Headers, Message, PublishReceipt, Stream};
6
7pub const DEAD_LETTER_REASON_MAX_DELIVERIES: &str = "max_deliveries_exceeded";
8
9#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
10pub struct DeliveryToken(pub u64);
11
12#[derive(Debug, Clone, PartialEq, Eq)]
13pub struct WorkItem {
14    pub token: DeliveryToken,
15    pub message: Message,
16    pub delivery_count: u32,
17}
18
19#[derive(Debug, Clone, PartialEq, Eq)]
20pub struct MaxDeliveriesExceeded {
21    pub item: WorkItem,
22    pub max_deliveries: u32,
23}
24
25#[derive(Debug, Clone, PartialEq, Eq)]
26pub enum ClaimOutcome {
27    Item(WorkItem),
28    Empty,
29    MaxDeliveriesExceeded(MaxDeliveriesExceeded),
30}
31
32/// A claimant on a bounded work queue.
33///
34/// `ack`, `nak` and `term` return `Ok` only once the server has recorded the
35/// decision, so a claimant may exit straight after one without losing it.
36#[async_trait]
37pub trait WorkQueue: Send + Sync {
38    async fn claim(&self) -> BusResult<ClaimOutcome>;
39    async fn ack(&self, token: DeliveryToken) -> BusResult<()>;
40    async fn nak(&self, token: DeliveryToken, delay: Duration) -> BusResult<()>;
41    async fn term(&self, token: DeliveryToken) -> BusResult<()>;
42}
43
44#[derive(Debug, Clone, PartialEq, Eq)]
45pub struct DeadLetterRecord {
46    pub original_subject: String,
47    pub message_id: String,
48    pub digest: ContentDigest,
49    pub delivery_count: u32,
50    pub reason: String,
51}
52
53impl DeadLetterRecord {
54    pub fn from_exhaustion(exhausted: &MaxDeliveriesExceeded) -> Self {
55        Self {
56            original_subject: exhausted.item.message.subject.clone(),
57            message_id: exhausted.item.message.id.clone(),
58            digest: exhausted.item.message.digest,
59            delivery_count: exhausted.item.delivery_count,
60            reason: DEAD_LETTER_REASON_MAX_DELIVERIES.into(),
61        }
62    }
63
64    pub fn headers(&self) -> Headers {
65        Headers::from([
66            ("original-subject".into(), self.original_subject.clone()),
67            ("message-id".into(), self.message_id.clone()),
68            ("content-digest".into(), self.digest.to_string()),
69            ("delivery-count".into(), self.delivery_count.to_string()),
70            ("reason".into(), self.reason.clone()),
71        ])
72    }
73}
74
75pub fn effect_dead_subject(account: &str) -> String {
76    format!("ck.{account}.effect.dead")
77}
78
79/// Publishes a persistent dead-letter record before terminating the exhausted delivery.
80/// Returns the publish receipt, including duplicates, as described by [`Stream::publish`].
81/// A duplicate is also followed by termination; use a separate dead-letter stream
82/// so its message ids do not collide with the original work items.
83pub async fn terminally_dispose<S, Q>(
84    stream: &S,
85    queue: &Q,
86    account: &str,
87    exhausted: &MaxDeliveriesExceeded,
88    dead_letter_digest: ContentDigest,
89) -> BusResult<PublishReceipt>
90where
91    S: Stream,
92    Q: WorkQueue,
93{
94    let record = DeadLetterRecord::from_exhaustion(exhausted);
95    let receipt = stream
96        .publish(
97            &effect_dead_subject(account),
98            &record.message_id,
99            dead_letter_digest,
100            record.headers(),
101        )
102        .await?;
103    queue.term(exhausted.item.token).await?;
104    Ok(receipt)
105}
cortexkit_bus_trait/work_queue.rs

cortexkit_bus_trait/
work_queue.rs
