awaken_stores/
pending_message_store.rs

1use async_trait::async_trait;
2use awaken_server_contract::contract::message::{
3    DeliveryBoundary, DeliveryMode, Message, MessageRecord, PendingMessageRecord,
4};
5use awaken_server_contract::contract::storage::{
6    RunRecord, StorageError, ThreadRunStore, message_append,
7};
8
9pub(crate) fn validate_pending_message_record(
10    record: &PendingMessageRecord,
11) -> Result<(), StorageError> {
12    if record.pending_id.trim().is_empty() {
13        return Err(StorageError::Validation(
14            "pending message id must not be empty".to_string(),
15        ));
16    }
17    if record.thread_id.trim().is_empty() {
18        return Err(StorageError::Validation(format!(
19            "pending message '{}' thread id must not be empty",
20            record.pending_id
21        )));
22    }
23    if record.position == 0 {
24        return Err(StorageError::Validation(format!(
25            "pending message '{}' position must be 1-based",
26            record.pending_id
27        )));
28    }
29    if record.revision == 0 {
30        return Err(StorageError::Validation(format!(
31            "pending message '{}' revision must be 1-based",
32            record.pending_id
33        )));
34    }
35    match record.message.id.as_deref() {
36        Some(message_id) if message_id == record.pending_id => {}
37        Some(message_id) => {
38            return Err(StorageError::Validation(format!(
39                "pending message '{}' cannot carry message id '{}'",
40                record.pending_id, message_id
41            )));
42        }
43        None => {
44            return Err(StorageError::Validation(format!(
45                "pending message '{}' must carry message id",
46                record.pending_id
47            )));
48        }
49    }
50    message_append::validate_message_shape(&record.message, "pending")
51}
52
53pub(crate) fn validate_pending_message_records(
54    records: &[PendingMessageRecord],
55) -> Result<(), StorageError> {
56    for record in records {
57        validate_pending_message_record(record)?;
58    }
59    Ok(())
60}
61
62/// Store-local extension for delivered-but-unconsumed thread messages.
63#[async_trait]
64pub trait PendingMessageStore: Send + Sync {
65    async fn load_pending_message_records(
66        &self,
67        thread_id: &str,
68    ) -> Result<Vec<PendingMessageRecord>, StorageError>;
69
70    /// List up to `limit` thread ids (ascending; `limit == 0` means unbounded)
71    /// that currently hold at least one pending message, strictly greater than
72    /// `after` (the previous page's last id) for cursor pagination. Startup
73    /// recovery pages through this to detect threads whose consume opportunity
74    /// may have been lost — pending was persisted but the dispatch/notification
75    /// did not survive — without scanning the whole table at once (ADR-0042 D7).
76    async fn list_threads_with_pending_messages(
77        &self,
78        limit: usize,
79        after: Option<&str>,
80    ) -> Result<Vec<String>, StorageError>;
81
82    async fn append_pending_message_records(
83        &self,
84        thread_id: &str,
85        messages: &[Message],
86        delivery_mode: DeliveryMode,
87    ) -> Result<Vec<PendingMessageRecord>, StorageError>;
88
89    async fn update_pending_message_record(
90        &self,
91        thread_id: &str,
92        pending_id: &str,
93        message: Message,
94    ) -> Result<PendingMessageRecord, StorageError> {
95        self.update_pending_message_record_checked(thread_id, pending_id, None, message)
96            .await
97    }
98
99    async fn update_pending_message_record_checked(
100        &self,
101        thread_id: &str,
102        pending_id: &str,
103        expected_revision: Option<u64>,
104        message: Message,
105    ) -> Result<PendingMessageRecord, StorageError>;
106
107    async fn retract_pending_message_record(
108        &self,
109        thread_id: &str,
110        pending_id: &str,
111    ) -> Result<PendingMessageRecord, StorageError> {
112        self.retract_pending_message_record_checked(thread_id, pending_id, None)
113            .await
114    }
115
116    async fn retract_pending_message_record_checked(
117        &self,
118        thread_id: &str,
119        pending_id: &str,
120        expected_revision: Option<u64>,
121    ) -> Result<PendingMessageRecord, StorageError>;
122
123    async fn reorder_pending_message_records(
124        &self,
125        thread_id: &str,
126        ordered_pending_ids: &[String],
127    ) -> Result<Vec<PendingMessageRecord>, StorageError> {
128        self.reorder_pending_message_records_checked(thread_id, None, ordered_pending_ids)
129            .await
130    }
131
132    async fn reorder_pending_message_records_checked(
133        &self,
134        thread_id: &str,
135        expected_queue_revision: Option<u64>,
136        ordered_pending_ids: &[String],
137    ) -> Result<Vec<PendingMessageRecord>, StorageError>;
138
139    async fn freeze_pending_message_records(
140        &self,
141        thread_id: &str,
142        boundary: DeliveryBoundary,
143        expected_message_version: Option<u64>,
144    ) -> Result<Vec<MessageRecord>, StorageError>;
145
146    async fn freeze_pending_message_records_with_run(
147        &self,
148        thread_id: &str,
149        boundary: DeliveryBoundary,
150        expected_message_version: Option<u64>,
151        expected_pending_ids: &[String],
152        run: &RunRecord,
153    ) -> Result<Vec<MessageRecord>, StorageError>;
154
155    /// Atomically append `new_messages` to pending and freeze the selected
156    /// pending entries (existing + newly appended) with the run record, in one
157    /// backend boundary (ADR-0042 D7).
158    ///
159    /// Closes the crash window a separate append-then-freeze leaves: a crash
160    /// between them persists pending with no consume context. Here a crash
161    /// either persists nothing (the client retry is the only request — no
162    /// duplicate) or the complete frozen run (no orphan). `expected_pending_ids`
163    /// is the caller's selection over `existing_pending ++ new_messages` (each
164    /// appended entry takes `pending_id == message id`); it is CAS-checked, so a
165    /// concurrent change since the caller's read aborts with
166    /// `PendingSelectionConflict`.
167    #[allow(clippy::too_many_arguments)]
168    async fn append_and_freeze_pending_message_records_with_run(
169        &self,
170        thread_id: &str,
171        new_messages: &[Message],
172        append_delivery_mode: DeliveryMode,
173        boundary: DeliveryBoundary,
174        expected_message_version: Option<u64>,
175        expected_pending_ids: &[String],
176        run: &RunRecord,
177    ) -> Result<Vec<MessageRecord>, StorageError>;
178}
179
180/// Thread/run store that owns the pending partition for the same backend.
181///
182/// ADR-0042 freeze operations consume pending messages and write committed
183/// messages plus the run record in one backend boundary, so mailbox wiring
184/// should depend on this combined capability instead of a separate pending
185/// store handle.
186pub trait PendingThreadRunStore: ThreadRunStore + PendingMessageStore {}
187
188impl<T> PendingThreadRunStore for T where T: ThreadRunStore + PendingMessageStore {}
awaken_stores/pending_message_store.rs

awaken_stores/
pending_message_store.rs
