Vakyartha documentation

1//! Append-preserving delivery outbox. Jobs are persisted before transport.
2
3use crate::{DeliveryJob, DeliveryPacket};
4use serde::{Deserialize, Serialize};
5use std::fs::{self, OpenOptions};
6use std::io::Write;
7use std::path::{Path, PathBuf};
8use std::sync::{Arc, Mutex, PoisonError};
9use std::time::{SystemTime, UNIX_EPOCH};
10
11#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
12pub struct OutboxRecord {
13 pub schema_version: u16,
14 pub job: DeliveryJob,
15 pub state: OutboxState,
16 pub attempts: u32,
17 pub created_at_ms: u64,
18 pub updated_at_ms: u64,
19 pub packet: Option<DeliveryPacket>,
20 pub last_error: Option<String>,
21}
22
23#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
24#[serde(rename_all = "snake_case")]
25pub enum OutboxState {
26 Pending,
27 Delivered,
28 DeadLetter,
29}
30
31#[derive(Debug)]
32pub enum OutboxError {
33 Io(String),
34 InvalidRecord(String),
35 Conflict(String),
36}
37
38impl std::fmt::Display for OutboxError {
39 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
40 match self {
41 Self::Io(message) => write!(formatter, "delivery outbox I/O failed: {message}"),
42 Self::InvalidRecord(message) => write!(formatter, "invalid outbox record: {message}"),
43 Self::Conflict(message) => write!(formatter, "delivery outbox conflict: {message}"),
44 }
45 }
46}
47
48impl std::error::Error for OutboxError {}
49
50#[derive(Debug, Clone)]
51pub struct Outbox {
52 root: PathBuf,
53 // Guards `update`'s read-modify-append against concurrent callers within
54 // this process. `update` has no file-level lock of its own, so without
55 // this, correctness would rest entirely on every caller happening to
56 // serialize through some *other* shared mutex (as `DeliveryRuntime`'s
57 // `serial` field does today) — an invariant invisible from this module
58 // and easy for a future call site to violate, reintroducing a classic
59 // lost-update race (two updates read the same `attempts`/state, the
60 // second overwrites the first). `Arc` so every `Outbox` clone sharing
61 // this `root` also shares the lock, not just the original instance.
62 // This still does not protect against a second *process* writing the
63 // same `root` concurrently — that would need a real file lock (e.g.
64 // flock), which nothing in this workspace does today because there is
65 // exactly one `Outbox` per `root` per process in practice.
66 lock: Arc<Mutex<()>>,
67}
68
69impl Outbox {
70 pub fn new(root: impl Into<PathBuf>) -> Self {
71 Self {
72 root: root.into(),
73 lock: Arc::new(Mutex::new(())),
74 }
75 }
76
77 pub fn enqueue(&self, job: DeliveryJob) -> Result<OutboxRecord, OutboxError> {
78 fs::create_dir_all(&self.root).map_err(io_error)?;
79 let path = self.record_path(&job.job_id);
80 if path.exists() {
81 let existing = read_record(&path)?;
82 if existing.job == job {
83 return Ok(existing);
84 }
85 return Err(OutboxError::Conflict(format!(
86 "job id {} already exists with different content",
87 job.job_id
88 )));
89 }
90 let now = epoch_millis();
91 let record = OutboxRecord {
92 schema_version: 1,
93 job,
94 state: OutboxState::Pending,
95 attempts: 0,
96 created_at_ms: now,
97 updated_at_ms: now,
98 packet: None,
99 last_error: None,
100 };
101 create_record(&path, &record)?;
102 Ok(record)
103 }
104
105 pub fn pending(&self) -> Result<Vec<OutboxRecord>, OutboxError> {
106 let mut records = self.list_filtered(|record| record.state == OutboxState::Pending)?;
107 // Replay is oldest-first so a busy outbox cannot starve its earliest
108 // durable job behind a stream of newer alerts.
109 records.sort_by_key(|record| (record.created_at_ms, record.job.job_id.clone()));
110 Ok(records)
111 }
112
113 /// Read every persisted delivery record, newest updates first. The
114 /// operations console uses this for evidence and replay; the source of
115 /// truth remains the append-preserving JSON files.
116 pub fn list(&self) -> Result<Vec<OutboxRecord>, OutboxError> {
117 self.list_filtered(|_| true)
118 }
119
120 pub fn get(&self, job_id: &str) -> Result<OutboxRecord, OutboxError> {
121 read_record(&self.record_path(job_id))
122 }
123
124 fn list_filtered(
125 &self,
126 include: impl Fn(&OutboxRecord) -> bool,
127 ) -> Result<Vec<OutboxRecord>, OutboxError> {
128 if !self.root.exists() {
129 return Ok(Vec::new());
130 }
131 let mut records = Vec::new();
132 for entry in fs::read_dir(&self.root).map_err(io_error)? {
133 let entry = entry.map_err(io_error)?;
134 let path = entry.path();
135 if path.extension().and_then(|value| value.to_str()) != Some("json") {
136 continue;
137 }
138 let record = read_record(&path)?;
139 if include(&record) {
140 records.push(record);
141 }
142 }
143 records.sort_by_key(|record| (record.updated_at_ms, record.job.job_id.clone()));
144 records.reverse();
145 Ok(records)
146 }
147
148 pub fn mark_delivered(
149 &self,
150 job_id: &str,
151 packet: DeliveryPacket,
152 ) -> Result<OutboxRecord, OutboxError> {
153 self.update(job_id, |record| {
154 record.state = OutboxState::Delivered;
155 record.attempts = record.attempts.saturating_add(1);
156 record.packet = Some(packet);
157 record.last_error = None;
158 })
159 }
160
161 pub fn mark_failed(
162 &self,
163 job_id: &str,
164 error: impl Into<String>,
165 ) -> Result<OutboxRecord, OutboxError> {
166 let error = error.into();
167 self.update(job_id, |record| {
168 record.attempts = record.attempts.saturating_add(1);
169 record.last_error = Some(error);
170 })
171 }
172
173 pub fn mark_dead_letter(
174 &self,
175 job_id: &str,
176 error: impl Into<String>,
177 ) -> Result<OutboxRecord, OutboxError> {
178 let error = error.into();
179 self.update(job_id, |record| {
180 record.state = OutboxState::DeadLetter;
181 record.attempts = record.attempts.saturating_add(1);
182 record.last_error = Some(error);
183 })
184 }
185
186 fn update(
187 &self,
188 job_id: &str,
189 mutate: impl FnOnce(&mut OutboxRecord),
190 ) -> Result<OutboxRecord, OutboxError> {
191 // Serialize the whole read-modify-append so two concurrent updates
192 // (e.g. a replay tick and a direct `deliver` call racing on the
193 // same job) can't both read the pre-mutation record and have the
194 // second overwrite the first's change. See the `lock` field doc.
195 let _guard = self.lock.lock().unwrap_or_else(PoisonError::into_inner);
196 let path = self.record_path(job_id);
197 let mut record = read_record(&path)?;
198 mutate(&mut record);
199 record.updated_at_ms = epoch_millis();
200 append_record(&path, &record)?;
201 Ok(record)
202 }
203
204 fn record_path(&self, job_id: &str) -> PathBuf {
205 self.root.join(format!("{}.json", hex_name(job_id)))
206 }
207}
208
209fn read_record(path: &Path) -> Result<OutboxRecord, OutboxError> {
210 let bytes = fs::read(path).map_err(io_error)?;
211 let record: OutboxRecord = match serde_json::from_slice(&bytes) {
212 Ok(record) => record,
213 Err(full_error) => bytes
214 .split(|byte| *byte == b'\n')
215 .rev()
216 .filter(|line| !line.is_empty())
217 .find_map(|line| serde_json::from_slice(line).ok())
218 .ok_or_else(|| OutboxError::InvalidRecord(full_error.to_string()))?,
219 };
220 if record.schema_version != 1 {
221 return Err(OutboxError::InvalidRecord(format!(
222 "unsupported schema version {}",
223 record.schema_version
224 )));
225 }
226 Ok(record)
227}
228
229fn create_record(path: &Path, record: &OutboxRecord) -> Result<(), OutboxError> {
230 let bytes = record_line(record)?;
231 let parent = path
232 .parent()
233 .ok_or_else(|| OutboxError::Io("record path has no parent".into()))?;
234 fs::create_dir_all(parent).map_err(io_error)?;
235 let mut file = OpenOptions::new()
236 .write(true)
237 .create_new(true)
238 .open(path)
239 .map_err(io_error)?;
240 file.write_all(&bytes).map_err(io_error)?;
241 file.sync_all().map_err(io_error)?;
242 sync_directory(parent)
243}
244
245fn append_record(path: &Path, record: &OutboxRecord) -> Result<(), OutboxError> {
246 let bytes = record_line(record)?;
247 let mut file = OpenOptions::new()
248 .append(true)
249 .open(path)
250 .map_err(io_error)?;
251 file.write_all(&bytes).map_err(io_error)?;
252 file.sync_all().map_err(io_error)
253}
254
255fn record_line(record: &OutboxRecord) -> Result<Vec<u8>, OutboxError> {
256 let mut bytes = serde_json::to_vec(record)
257 .map_err(|error| OutboxError::InvalidRecord(error.to_string()))?;
258 bytes.push(b'\n');
259 Ok(bytes)
260}
261
262#[cfg(unix)]
263fn sync_directory(path: &Path) -> Result<(), OutboxError> {
264 fs::File::open(path)
265 .and_then(|directory| directory.sync_all())
266 .map_err(io_error)
267}
268
269#[cfg(not(unix))]
270fn sync_directory(_path: &Path) -> Result<(), OutboxError> {
271 Ok(())
272}
273
274fn hex_name(value: &str) -> String {
275 value
276 .as_bytes()
277 .iter()
278 .map(|byte| format!("{byte:02x}"))
279 .collect()
280}
281
282fn epoch_millis() -> u64 {
283 SystemTime::now()
284 .duration_since(UNIX_EPOCH)
285 .map(|duration| duration.as_millis().min(u128::from(u64::MAX)) as u64)
286 .unwrap_or_default()
287}
288
289fn io_error(error: std::io::Error) -> OutboxError {
290 OutboxError::Io(error.to_string())
291}
292
293#[cfg(test)]
294#[allow(clippy::expect_used)]
295mod tests {
296 use super::*;
297 use crate::{AnswerDraft, DeliveryContent, DeliveryKind, DeliveryProfile};
298
299 fn test_job() -> DeliveryJob {
300 DeliveryJob {
301 job_id: "job/one".into(),
302 target: "log:test".into(),
303 kind: DeliveryKind::Assistant,
304 content: DeliveryContent::Answer(AnswerDraft::from_markdown("exact **answer**")),
305 profile: DeliveryProfile::plain("log"),
306 skill_registry: None,
307 }
308 }
309
310 #[test]
311 fn preserves_job_until_delivered() {
312 let root = std::env::temp_dir().join(format!(
313 "vak-delivery-outbox-{}-{}",
314 std::process::id(),
315 epoch_millis()
316 ));
317 let outbox = Outbox::new(&root);
318 let job = test_job();
319 outbox.enqueue(job.clone()).expect("enqueue");
320 assert_eq!(outbox.pending().expect("pending")[0].job, job);
321 let packet = crate::render(&job).expect("render");
322 outbox.mark_delivered(&job.job_id, packet).expect("mark");
323 assert!(outbox.pending().expect("pending").is_empty());
324 let record_path = std::fs::read_dir(&root)
325 .expect("records")
326 .next()
327 .expect("record entry")
328 .expect("record path")
329 .path();
330 let history = std::fs::read_to_string(&record_path).expect("history");
331 assert_eq!(history.lines().count(), 2);
332 assert!(history.contains("exact **answer**"));
333 std::fs::OpenOptions::new()
334 .append(true)
335 .open(record_path)
336 .expect("open partial")
337 .write_all(b"{\"partial\":")
338 .expect("write partial");
339 assert!(
340 outbox.pending().expect("pending after partial").is_empty(),
341 "a torn final append must not hide the last valid state"
342 );
343 let _ = std::fs::remove_dir_all(root);
344 }
345}
346

1//! Append-preserving delivery outbox. Jobs are persisted before transport.
2
3use crate::{DeliveryJob, DeliveryPacket};
4use serde::{Deserialize, Serialize};
5use std::fs::{self, OpenOptions};
6use std::io::Write;
7use std::path::{Path, PathBuf};
8use std::sync::{Arc, Mutex, PoisonError};
9use std::time::{SystemTime, UNIX_EPOCH};
10
11#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
12pub struct OutboxRecord {
13 pub schema_version: u16,
14 pub job: DeliveryJob,
15 pub state: OutboxState,
16 pub attempts: u32,
17 pub created_at_ms: u64,
18 pub updated_at_ms: u64,
19 pub packet: Option<DeliveryPacket>,
20 pub last_error: Option<String>,
21}
22
23#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
24#[serde(rename_all = "snake_case")]
25pub enum OutboxState {
26 Pending,
27 Delivered,
28 DeadLetter,
29}
30
31#[derive(Debug)]
32pub enum OutboxError {
33 Io(String),
34 InvalidRecord(String),
35 Conflict(String),
36}
37
38impl std::fmt::Display for OutboxError {
39 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
40 match self {
41 Self::Io(message) => write!(formatter, "delivery outbox I/O failed: {message}"),
42 Self::InvalidRecord(message) => write!(formatter, "invalid outbox record: {message}"),
43 Self::Conflict(message) => write!(formatter, "delivery outbox conflict: {message}"),
44 }
45 }
46}
47
48impl std::error::Error for OutboxError {}
49
50#[derive(Debug, Clone)]
51pub struct Outbox {
52 root: PathBuf,
53 // Guards `update`'s read-modify-append against concurrent callers within
54 // this process. `update` has no file-level lock of its own, so without
55 // this, correctness would rest entirely on every caller happening to
56 // serialize through some *other* shared mutex (as `DeliveryRuntime`'s
57 // `serial` field does today) — an invariant invisible from this module
58 // and easy for a future call site to violate, reintroducing a classic
59 // lost-update race (two updates read the same `attempts`/state, the
60 // second overwrites the first). `Arc` so every `Outbox` clone sharing
61 // this `root` also shares the lock, not just the original instance.
62 // This still does not protect against a second *process* writing the
63 // same `root` concurrently — that would need a real file lock (e.g.
64 // flock), which nothing in this workspace does today because there is
65 // exactly one `Outbox` per `root` per process in practice.
66 lock: Arc<Mutex<()>>,
67}
68
69impl Outbox {
70 pub fn new(root: impl Into<PathBuf>) -> Self {
71 Self {
72 root: root.into(),
73 lock: Arc::new(Mutex::new(())),
74 }
75 }
76
77 pub fn enqueue(&self, job: DeliveryJob) -> Result<OutboxRecord, OutboxError> {
78 fs::create_dir_all(&self.root).map_err(io_error)?;
79 let path = self.record_path(&job.job_id);
80 if path.exists() {
81 let existing = read_record(&path)?;
82 if existing.job == job {
83 return Ok(existing);
84 }
85 return Err(OutboxError::Conflict(format!(
86 "job id {} already exists with different content",
87 job.job_id
88 )));
89 }
90 let now = epoch_millis();
91 let record = OutboxRecord {
92 schema_version: 1,
93 job,
94 state: OutboxState::Pending,
95 attempts: 0,
96 created_at_ms: now,
97 updated_at_ms: now,
98 packet: None,
99 last_error: None,
100 };
101 create_record(&path, &record)?;
102 Ok(record)
103 }
104
105 pub fn pending(&self) -> Result<Vec<OutboxRecord>, OutboxError> {
106 let mut records = self.list_filtered(|record| record.state == OutboxState::Pending)?;
107 // Replay is oldest-first so a busy outbox cannot starve its earliest
108 // durable job behind a stream of newer alerts.
109 records.sort_by_key(|record| (record.created_at_ms, record.job.job_id.clone()));
110 Ok(records)
111 }
112
113 /// Read every persisted delivery record, newest updates first. The
114 /// operations console uses this for evidence and replay; the source of
115 /// truth remains the append-preserving JSON files.
116 pub fn list(&self) -> Result<Vec<OutboxRecord>, OutboxError> {
117 self.list_filtered(|_| true)
118 }
119
120 pub fn get(&self, job_id: &str) -> Result<OutboxRecord, OutboxError> {
121 read_record(&self.record_path(job_id))
122 }
123
124 fn list_filtered(
125 &self,
126 include: impl Fn(&OutboxRecord) -> bool,
127 ) -> Result<Vec<OutboxRecord>, OutboxError> {
128 if !self.root.exists() {
129 return Ok(Vec::new());
130 }
131 let mut records = Vec::new();
132 for entry in fs::read_dir(&self.root).map_err(io_error)? {
133 let entry = entry.map_err(io_error)?;
134 let path = entry.path();
135 if path.extension().and_then(|value| value.to_str()) != Some("json") {
136 continue;
137 }
138 let record = read_record(&path)?;
139 if include(&record) {
140 records.push(record);
141 }
142 }
143 records.sort_by_key(|record| (record.updated_at_ms, record.job.job_id.clone()));
144 records.reverse();
145 Ok(records)
146 }
147
148 pub fn mark_delivered(
149 &self,
150 job_id: &str,
151 packet: DeliveryPacket,
152 ) -> Result<OutboxRecord, OutboxError> {
153 self.update(job_id, |record| {
154 record.state = OutboxState::Delivered;
155 record.attempts = record.attempts.saturating_add(1);
156 record.packet = Some(packet);
157 record.last_error = None;
158 })
159 }
160
161 pub fn mark_failed(
162 &self,
163 job_id: &str,
164 error: impl Into<String>,
165 ) -> Result<OutboxRecord, OutboxError> {
166 let error = error.into();
167 self.update(job_id, |record| {
168 record.attempts = record.attempts.saturating_add(1);
169 record.last_error = Some(error);
170 })
171 }
172
173 pub fn mark_dead_letter(
174 &self,
175 job_id: &str,
176 error: impl Into<String>,
177 ) -> Result<OutboxRecord, OutboxError> {
178 let error = error.into();
179 self.update(job_id, |record| {
180 record.state = OutboxState::DeadLetter;
181 record.attempts = record.attempts.saturating_add(1);
182 record.last_error = Some(error);
183 })
184 }
185
186 fn update(
187 &self,
188 job_id: &str,
189 mutate: impl FnOnce(&mut OutboxRecord),
190 ) -> Result<OutboxRecord, OutboxError> {
191 // Serialize the whole read-modify-append so two concurrent updates
192 // (e.g. a replay tick and a direct `deliver` call racing on the
193 // same job) can't both read the pre-mutation record and have the
194 // second overwrite the first's change. See the `lock` field doc.
195 let _guard = self.lock.lock().unwrap_or_else(PoisonError::into_inner);
196 let path = self.record_path(job_id);
197 let mut record = read_record(&path)?;
198 mutate(&mut record);
199 record.updated_at_ms = epoch_millis();
200 append_record(&path, &record)?;
201 Ok(record)
202 }
203
204 fn record_path(&self, job_id: &str) -> PathBuf {
205 self.root.join(format!("{}.json", hex_name(job_id)))
206 }
207}
208
209fn read_record(path: &Path) -> Result<OutboxRecord, OutboxError> {
210 let bytes = fs::read(path).map_err(io_error)?;
211 let record: OutboxRecord = match serde_json::from_slice(&bytes) {
212 Ok(record) => record,
213 Err(full_error) => bytes
214 .split(|byte| *byte == b'\n')
215 .rev()
216 .filter(|line| !line.is_empty())
217 .find_map(|line| serde_json::from_slice(line).ok())
218 .ok_or_else(|| OutboxError::InvalidRecord(full_error.to_string()))?,
219 };
220 if record.schema_version != 1 {
221 return Err(OutboxError::InvalidRecord(format!(
222 "unsupported schema version {}",
223 record.schema_version
224 )));
225 }
226 Ok(record)
227}
228
229fn create_record(path: &Path, record: &OutboxRecord) -> Result<(), OutboxError> {
230 let bytes = record_line(record)?;
231 let parent = path
232 .parent()
233 .ok_or_else(|| OutboxError::Io("record path has no parent".into()))?;
234 fs::create_dir_all(parent).map_err(io_error)?;
235 let mut file = OpenOptions::new()
236 .write(true)
237 .create_new(true)
238 .open(path)
239 .map_err(io_error)?;
240 file.write_all(&bytes).map_err(io_error)?;
241 file.sync_all().map_err(io_error)?;
242 sync_directory(parent)
243}
244
245fn append_record(path: &Path, record: &OutboxRecord) -> Result<(), OutboxError> {
246 let bytes = record_line(record)?;
247 let mut file = OpenOptions::new()
248 .append(true)
249 .open(path)
250 .map_err(io_error)?;
251 file.write_all(&bytes).map_err(io_error)?;
252 file.sync_all().map_err(io_error)
253}
254
255fn record_line(record: &OutboxRecord) -> Result<Vec<u8>, OutboxError> {
256 let mut bytes = serde_json::to_vec(record)
257 .map_err(|error| OutboxError::InvalidRecord(error.to_string()))?;
258 bytes.push(b'\n');
259 Ok(bytes)
260}
261
262#[cfg(unix)]
263fn sync_directory(path: &Path) -> Result<(), OutboxError> {
264 fs::File::open(path)
265 .and_then(|directory| directory.sync_all())
266 .map_err(io_error)
267}
268
269#[cfg(not(unix))]
270fn sync_directory(_path: &Path) -> Result<(), OutboxError> {
271 Ok(())
272}
273
274fn hex_name(value: &str) -> String {
275 value
276 .as_bytes()
277 .iter()
278 .map(|byte| format!("{byte:02x}"))
279 .collect()
280}
281
282fn epoch_millis() -> u64 {
283 SystemTime::now()
284 .duration_since(UNIX_EPOCH)
285 .map(|duration| duration.as_millis().min(u128::from(u64::MAX)) as u64)
286 .unwrap_or_default()
287}
288
289fn io_error(error: std::io::Error) -> OutboxError {
290 OutboxError::Io(error.to_string())
291}
292
293#[cfg(test)]
294#[allow(clippy::expect_used)]
295mod tests {
296 use super::*;
297 use crate::{AnswerDraft, DeliveryContent, DeliveryKind, DeliveryProfile};
298
299 fn test_job() -> DeliveryJob {
300 DeliveryJob {
301 job_id: "job/one".into(),
302 target: "log:test".into(),
303 kind: DeliveryKind::Assistant,
304 content: DeliveryContent::Answer(AnswerDraft::from_markdown("exact **answer**")),
305 profile: DeliveryProfile::plain("log"),
306 skill_registry: None,
307 }
308 }
309
310 #[test]
311 fn preserves_job_until_delivered() {
312 let root = std::env::temp_dir().join(format!(
313 "vak-delivery-outbox-{}-{}",
314 std::process::id(),
315 epoch_millis()
316 ));
317 let outbox = Outbox::new(&root);
318 let job = test_job();
319 outbox.enqueue(job.clone()).expect("enqueue");
320 assert_eq!(outbox.pending().expect("pending")[0].job, job);
321 let packet = crate::render(&job).expect("render");
322 outbox.mark_delivered(&job.job_id, packet).expect("mark");
323 assert!(outbox.pending().expect("pending").is_empty());
324 let record_path = std::fs::read_dir(&root)
325 .expect("records")
326 .next()
327 .expect("record entry")
328 .expect("record path")
329 .path();
330 let history = std::fs::read_to_string(&record_path).expect("history");
331 assert_eq!(history.lines().count(), 2);
332 assert!(history.contains("exact **answer**"));
333 std::fs::OpenOptions::new()
334 .append(true)
335 .open(record_path)
336 .expect("open partial")
337 .write_all(b"{\"partial\":")
338 .expect("write partial");
339 assert!(
340 outbox.pending().expect("pending after partial").is_empty(),
341 "a torn final append must not hide the last valid state"
342 );
343 let _ = std::fs::remove_dir_all(root);
344 }
345}
346

Indexing the workspace…