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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
//! v0.30 / PR α' — file-based outbox for `EventStarted` publishes.
//!
//! Mirrors `crate::outbox`'s ExecResult design: atomic tmp-then-
//! rename enqueue + a background drain task that publishes via
//! JetStream PubAck, deleting the file only on ack. Separate
//! directory (`events-outbox/`) so existing v0.24 ExecResult outbox
//! files keep working unchanged across the upgrade.
//!
//! Why a parallel outbox instead of one generic envelope: lets the
//! existing on-disk format keep its meaning literally (= `<rid>.json`
//! IS an ExecResult, no envelope wrapping needed), at the cost of
//! ~80 lines of duplication. Avoids a one-shot file-format
//! migration on startup.
//!
//! Offline-safety: an agent that goes offline mid-Command still has
//! its lifecycle events reach the backend on reconnect. Without
//! persistence, a process restart between "spawn" and "publish"
//! would lose the started event entirely, and the matching
//! ExecResult arriving later would create the execution_results
//! row with `started_at` from the result side and no version/manifest
//! propagation from the start side — survivable but incomplete.
use std::path::{Path, PathBuf};
use std::time::Duration;
use anyhow::{Context, Result};
use kanade_shared::subject;
use kanade_shared::wire::EventStarted;
use tracing::{debug, warn};
const DRAIN_INTERVAL: Duration = Duration::from_secs(1);
/// Hard upper bound on how long we'll wait for a single publish's
/// PubAck before giving up on that file for this drain iteration.
/// Matches `outbox::ACK_TIMEOUT`; see that constant's doc for the
/// rationale (#139).
const ACK_TIMEOUT: Duration = Duration::from_secs(30);
/// Atomically persist one `EventStarted` to `events_outbox_dir`.
/// Filename is `<result_id>.json` — guaranteed unique per (exec,
/// pc) run because the agent mints `result_id` once per
/// handle_command invocation.
pub fn enqueue(events_outbox_dir: &Path, event: &EventStarted) -> Result<PathBuf> {
std::fs::create_dir_all(events_outbox_dir)
.with_context(|| format!("create events outbox dir {events_outbox_dir:?}"))?;
let final_path = events_outbox_dir.join(format!("{}.json", event.result_id));
let tmp_path = events_outbox_dir.join(format!("{}.json.tmp", event.result_id));
let bytes = serde_json::to_vec(event).context("serialise EventStarted")?;
std::fs::write(&tmp_path, &bytes)
.with_context(|| format!("write tmp events outbox file {tmp_path:?}"))?;
std::fs::rename(&tmp_path, &final_path)
.with_context(|| format!("rename tmp → {final_path:?}"))?;
Ok(final_path)
}
/// Long-running drain task. Each iteration scans the directory,
/// publishes pending events via `js.publish().await.await`
/// (PubAck-waited), deletes on success. Mirrors
/// `outbox::spawn_drain`.
pub fn spawn_drain(
client: async_nats::Client,
events_outbox_dir: PathBuf,
) -> tokio::task::JoinHandle<()> {
tokio::spawn(async move {
let js = async_nats::jetstream::new(client);
// Gemini #73 high fix: retry mkdir on each loop iteration
// instead of dying forever. Permission / disk-full / parent-
// missing failures are operationally recoverable, and dying
// here would silently swallow every subsequent enqueue.
// Per-file retry state, owned by the drain task so it survives across
// ticks. A ledger scoped to one `drain_once` call would reset every
// second and back off nothing.
let mut ledger = crate::outbox_retry::RetryLedger::new();
loop {
if let Err(e) = std::fs::create_dir_all(&events_outbox_dir) {
warn!(
error = %e,
dir = %events_outbox_dir.display(),
"events_outbox: create dir failed; will retry next tick",
);
tokio::time::sleep(DRAIN_INTERVAL).await;
continue;
}
drain_once(&js, &events_outbox_dir, &mut ledger).await;
tokio::time::sleep(DRAIN_INTERVAL).await;
}
})
}
async fn drain_once(
js: &async_nats::jetstream::Context,
events_outbox_dir: &Path,
ledger: &mut crate::outbox_retry::RetryLedger,
) {
let entries = match std::fs::read_dir(events_outbox_dir) {
Ok(e) => e,
Err(e) => {
warn!(
error = %e,
dir = %events_outbox_dir.display(),
"events_outbox: read_dir failed",
);
return;
}
};
let mut files: Vec<PathBuf> = entries
.filter_map(|r| r.ok().map(|e| e.path()))
.filter(|p| p.extension().is_some_and(|e| e == "json"))
.collect();
if files.is_empty() {
return;
}
// mtime-based ordering — `sort_by_cached_key` reads each file's
// metadata once instead of in every comparator step. Negligible
// at low file counts but matters when a long broker outage left
// hundreds of pending events.
files.sort_by_cached_key(|p| std::fs::metadata(p).and_then(|m| m.modified()).ok());
// Gemini #73 high fix: don't `return` on a single file's
// failure — continue to subsequent files. The original
// "stop on first error" was meant as backpressure for the
// common broker-down case (every publish_one would fail).
// Trade: a broker-down sweep now tries every file and logs
// each (debug level, low noise) before sleeping. Upside: a
// single problematic file (transient NATS error specific to
// its payload, ack timeout, etc.) no longer pins the entire
// outbox behind it. The remaining unpublished files stay on
// disk for the next tick to retry.
// Forget files that have left the directory, so the ledger tracks the
// currently-stuck set rather than everything ever seen (#1319).
ledger.retain_present(&files);
for path in files {
// Read per file, not once per sweep. Each `publish_one` can block up
// to `ACK_TIMEOUT`, so a sweep over a long queue outlasts the drain
// interval by minutes — anchoring every decision to the instant the
// sweep began would shorten the real backoff and skip files that
// became due while it ran.
let now = std::time::Instant::now();
// Checked BEFORE the read, not after: `publish_one` opens with
// `std::fs::read` of the whole file, and re-reading the queue every
// second is what pinned a core for two days. Skipping here is the
// entire saving.
if !ledger.is_due(&path, now) {
continue;
}
match publish_one(js, &path).await {
Ok(()) => ledger.record_success(&path),
Err(e) => {
let outcome = ledger.record_failure(&path, now);
if ledger.should_warn(&path) {
warn!(
error = %e,
path = %path.display(),
failures = ledger.failures(&path),
"events_outbox: publish keeps failing for this file",
);
} else {
debug!(
error = %e,
path = %path.display(),
"events_outbox: publish failed; backing off, continuing with others",
);
}
if outcome == crate::outbox_retry::AfterFailure::Quarantine
&& crate::outbox_retry::quarantine(&path, "events_outbox")
{
// Only once the file has actually left the directory.
// Clearing it for a file still sitting there would
// restart it at zero failures with no backoff — the 1 Hz
// re-read this change removes, back again.
ledger.record_success(&path);
}
}
}
}
}
async fn publish_one(js: &async_nats::jetstream::Context, path: &Path) -> Result<()> {
let bytes = std::fs::read(path).with_context(|| format!("read {path:?}"))?;
// Treat parse failure as "drop the corrupted file and continue"
// rather than propagating Err — drain_once bails on any error
// from publish_one, so a single corrupted file would otherwise
// wedge the whole outbox.
let event: EventStarted = match serde_json::from_slice(&bytes) {
Ok(e) => e,
Err(e) => {
warn!(
error = %e,
path = %path.display(),
"events_outbox: corrupted file — removing so drain can proceed",
);
let _ = std::fs::remove_file(path);
return Ok(());
}
};
let subj = subject::events_started(&event.exec_id, &event.pc_id);
let ack_future = js
.publish(subj.clone(), bytes.clone().into())
.await
.with_context(|| format!("publish {subj}"))?;
// Bounded ack wait (#139): a wedged broker or a stalled
// upstream stream must not pin the whole drain loop behind one
// file. On timeout the file stays on disk for the next drain
// iteration to retry.
let _ack = tokio::time::timeout(ACK_TIMEOUT, ack_future)
.await
.with_context(|| format!("ack timeout {subj} after {}s", ACK_TIMEOUT.as_secs()))?
.with_context(|| format!("ack {subj}"))?;
std::fs::remove_file(path).with_context(|| format!("remove {path:?}"))?;
debug!(
result_id = %event.result_id,
exec_id = %event.exec_id,
pc_id = %event.pc_id,
subject = %subj,
"events_outbox: started event delivered + file removed",
);
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use chrono::{TimeZone, Utc};
fn sample(result_id: &str, exec_id: &str, pc_id: &str) -> EventStarted {
EventStarted {
result_id: result_id.into(),
request_id: "req-1".into(),
exec_id: exec_id.into(),
pc_id: pc_id.into(),
started_at: Utc.with_ymd_and_hms(2026, 5, 20, 12, 0, 0).unwrap(),
manifest_id: "inv-hw".into(),
version: "1.0.0".into(),
}
}
#[test]
fn enqueue_creates_file_named_by_result_id() {
let dir = tempfile::tempdir().unwrap();
let e = sample("res-1", "exec-1", "pc-01");
let path = enqueue(dir.path(), &e).unwrap();
assert_eq!(path.file_name().unwrap(), "res-1.json");
let back: EventStarted = serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
assert_eq!(back.result_id, "res-1");
assert_eq!(back.exec_id, "exec-1");
}
#[test]
fn enqueue_atomic_overwrite() {
let dir = tempfile::tempdir().unwrap();
let e1 = sample("res-x", "exec-x", "pc-1");
let e2 = EventStarted {
manifest_id: "different".into(),
..sample("res-x", "exec-x", "pc-1")
};
enqueue(dir.path(), &e1).unwrap();
let path = enqueue(dir.path(), &e2).unwrap();
let back: EventStarted = serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
assert_eq!(back.manifest_id, "different");
}
}