1use anyhow::{bail, Context, Result};
2use rusqlite::{params, Connection, OptionalExtension, Transaction, TransactionBehavior};
3use serde::Serialize;
4
5use crate::{db, memory, workstream};
6
7const CLEANUP_POLICY_VERSION: i64 = 1;
8const FAILURE_ERROR_LIMIT_BYTES: usize = 1_000;
9
10#[derive(Debug, Clone, Copy, PartialEq, Eq)]
11pub struct CleanupPolicy {
12 archived_failure_days: Option<i64>,
13}
14
15impl CleanupPolicy {
16 pub fn automatic() -> Self {
17 Self {
18 archived_failure_days: None,
19 }
20 }
21
22 pub fn manual(archived_failure_days: Option<i64>) -> Result<Self> {
23 if archived_failure_days.is_some_and(|days| days <= 0) {
24 bail!("--archived-failures must be a positive number of days");
25 }
26 Ok(Self {
27 archived_failure_days,
28 })
29 }
30
31 pub fn retention_days(self) -> CleanupRetentionDays {
32 CleanupRetentionDays {
33 old_events: memory::OLD_EVENT_RETENTION_DAYS,
34 compressed_source_observations: memory::COMPRESSED_SOURCE_OBSERVATION_RETENTION_DAYS,
35 stale_memories: memory::STALE_MEMORY_ARCHIVE_DAYS,
36 archived_failures: self
37 .archived_failure_days
38 .unwrap_or(db::ARCHIVED_FAILURE_PURGE_DAYS),
39 workstream_auto_pause: workstream::DEFAULT_AUTO_PAUSE_DAYS,
40 workstream_auto_abandon: workstream::DEFAULT_AUTO_ABANDON_DAYS,
41 }
42 }
43}
44
45#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
46pub struct CleanupRetentionDays {
47 pub old_events: i64,
48 pub compressed_source_observations: i64,
49 pub stale_memories: i64,
50 pub archived_failures: i64,
51 pub workstream_auto_pause: i64,
52 pub workstream_auto_abandon: i64,
53}
54
55#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
56pub struct CleanupPlan {
57 pub expired_memories_to_stale: usize,
58 pub inactive_workstreams_to_pause: usize,
59 pub long_paused_workstreams_to_abandon: usize,
60 pub old_events_to_delete: usize,
61 pub compressed_source_observations_to_delete: usize,
62 pub stale_memories_to_archive: usize,
63 pub archived_failures_to_purge: db::ArchivedFailurePurgePlan,
64}
65
66#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
67pub struct CleanupApplied {
68 pub expired_memories_marked_stale: usize,
69 pub inactive_workstreams_paused: usize,
70 pub long_paused_workstreams_abandoned: usize,
71 pub old_events_deleted: usize,
72 pub compressed_source_observations_deleted: usize,
73 pub stale_memories_archived: usize,
74 pub archived_failures_purged: db::ArchivedFailurePurgePlan,
75}
76
77impl CleanupApplied {
78 fn matches_plan(&self, plan: &CleanupPlan) -> bool {
79 self.expired_memories_marked_stale == plan.expired_memories_to_stale
80 && self.inactive_workstreams_paused == plan.inactive_workstreams_to_pause
81 && self.long_paused_workstreams_abandoned == plan.long_paused_workstreams_to_abandon
82 && self.old_events_deleted == plan.old_events_to_delete
83 && self.compressed_source_observations_deleted
84 == plan.compressed_source_observations_to_delete
85 && self.stale_memories_archived == plan.stale_memories_to_archive
86 && self.archived_failures_purged == plan.archived_failures_to_purge
87 }
88}
89
90#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
91pub struct CleanupExecution {
92 pub plan: CleanupPlan,
93 pub applied: CleanupApplied,
94}
95
96#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
97pub struct CleanupReport {
98 pub dry_run: bool,
99 pub retention_days: CleanupRetentionDays,
100 pub plan: CleanupPlan,
101 pub applied: Option<CleanupApplied>,
102}
103
104#[derive(Debug, Clone, PartialEq, Eq)]
105pub struct CleanupRun {
106 pub id: i64,
107 pub job_id: Option<i64>,
108 pub started_at_epoch: i64,
109 pub finished_at_epoch: i64,
110 pub outcome: String,
111 pub counts_json: Option<String>,
112 pub error: Option<String>,
113}
114
115pub fn preview_cleanup(
116 conn: &Connection,
117 now_epoch: i64,
118 policy: CleanupPolicy,
119) -> Result<CleanupPlan> {
120 build_cleanup_plan(conn, now_epoch, policy)
121}
122
123pub fn execute_manual_cleanup(
124 conn: &Connection,
125 now_epoch: i64,
126 policy: CleanupPolicy,
127) -> Result<CleanupExecution> {
128 execute_cleanup(conn, now_epoch, policy, CleanupTrigger::Manual)
129}
130
131pub fn execute_automatic_cleanup_job(
132 conn: &Connection,
133 job_id: i64,
134 lease_owner: &str,
135 now_epoch: i64,
136) -> Result<CleanupExecution> {
137 execute_cleanup(
138 conn,
139 now_epoch,
140 CleanupPolicy::automatic(),
141 CleanupTrigger::Automatic {
142 job_id,
143 lease_owner,
144 },
145 )
146}
147
148pub fn latest_automatic_cleanup_run(
149 conn: &Connection,
150 outcome: &str,
151) -> Result<Option<CleanupRun>> {
152 if !matches!(outcome, "success" | "failure") {
153 bail!("cleanup run outcome must be success or failure");
154 }
155 conn.query_row(
156 "SELECT id, job_id, started_at_epoch, finished_at_epoch, outcome,
157 counts_json, error
158 FROM maintenance_runs
159 WHERE \"trigger\" = 'automatic' AND outcome = ?1
160 ORDER BY finished_at_epoch DESC, id DESC
161 LIMIT 1",
162 params![outcome],
163 |row| {
164 Ok(CleanupRun {
165 id: row.get(0)?,
166 job_id: row.get(1)?,
167 started_at_epoch: row.get(2)?,
168 finished_at_epoch: row.get(3)?,
169 outcome: row.get(4)?,
170 counts_json: row.get(5)?,
171 error: row.get(6)?,
172 })
173 },
174 )
175 .optional()
176 .context("read latest automatic cleanup run")
177}
178
179#[derive(Clone, Copy)]
180enum CleanupTrigger<'a> {
181 Manual,
182 Automatic { job_id: i64, lease_owner: &'a str },
183}
184
185impl CleanupTrigger<'_> {
186 fn as_str(self) -> &'static str {
187 match self {
188 Self::Manual => "manual",
189 Self::Automatic { .. } => "automatic",
190 }
191 }
192
193 fn job_id(self) -> Option<i64> {
194 match self {
195 Self::Manual => None,
196 Self::Automatic { job_id, .. } => Some(job_id),
197 }
198 }
199}
200
201fn execute_cleanup(
202 conn: &Connection,
203 now_epoch: i64,
204 policy: CleanupPolicy,
205 trigger: CleanupTrigger<'_>,
206) -> Result<CleanupExecution> {
207 let tx = match Transaction::new_unchecked(conn, TransactionBehavior::Immediate) {
208 Ok(tx) => tx,
209 Err(error) => {
210 let error = anyhow::Error::from(error).context("begin cleanup transaction");
211 return Err(record_failure_after_rollback(
212 conn, trigger, now_epoch, error,
213 ));
214 }
215 };
216 let execution = execute_cleanup_in_transaction(&tx, now_epoch, policy, trigger);
217 let execution = match execution {
218 Ok(execution) => execution,
219 Err(error) => {
220 drop(tx);
221 return Err(record_failure_after_rollback(
222 conn, trigger, now_epoch, error,
223 ));
224 }
225 };
226 if let Err(error) = tx.commit().context("commit cleanup transaction") {
227 return Err(record_failure_after_rollback(
228 conn, trigger, now_epoch, error,
229 ));
230 }
231 Ok(execution)
232}
233
234fn execute_cleanup_in_transaction(
235 conn: &Connection,
236 started_at_epoch: i64,
237 policy: CleanupPolicy,
238 trigger: CleanupTrigger<'_>,
239) -> Result<CleanupExecution> {
240 if let CleanupTrigger::Automatic {
241 job_id,
242 lease_owner,
243 } = trigger
244 {
245 validate_automatic_job(conn, job_id, lease_owner, started_at_epoch)?;
246 }
247 let plan = build_cleanup_plan(conn, started_at_epoch, policy)?;
248 let applied = apply_cleanup_plan(conn, started_at_epoch, policy)?;
249 if !applied.matches_plan(&plan) {
250 bail!("cleanup plan/apply count invariant failed");
251 }
252 let finished_at_epoch = cleanup_finished_at(started_at_epoch);
253 insert_success_run(conn, trigger, started_at_epoch, finished_at_epoch, &applied)?;
254 if let CleanupTrigger::Automatic {
255 job_id,
256 lease_owner,
257 } = trigger
258 {
259 finish_automatic_job(
260 conn,
261 job_id,
262 lease_owner,
263 started_at_epoch,
264 finished_at_epoch,
265 )?;
266 }
267 Ok(CleanupExecution { plan, applied })
268}
269
270fn build_cleanup_plan(
271 conn: &Connection,
272 now_epoch: i64,
273 policy: CleanupPolicy,
274) -> Result<CleanupPlan> {
275 Ok(CleanupPlan {
276 expired_memories_to_stale: memory::lifecycle::count_expired_active_memories(
277 conn, now_epoch,
278 )?,
279 inactive_workstreams_to_pause: workstream::count_auto_pause_all_inactive_at(
280 conn,
281 now_epoch,
282 workstream::DEFAULT_AUTO_PAUSE_DAYS,
283 )?,
284 long_paused_workstreams_to_abandon: workstream::count_auto_abandon_all_inactive_at(
285 conn,
286 now_epoch,
287 workstream::DEFAULT_AUTO_ABANDON_DAYS,
288 )?,
289 old_events_to_delete: memory::count_old_events_at(
290 conn,
291 now_epoch,
292 memory::OLD_EVENT_RETENTION_DAYS,
293 )?,
294 compressed_source_observations_to_delete:
295 memory::count_compressed_source_observations_to_delete_at(
296 conn,
297 now_epoch,
298 memory::COMPRESSED_SOURCE_OBSERVATION_RETENTION_DAYS,
299 )?,
300 stale_memories_to_archive: memory::count_stale_memories_to_archive_at(
301 conn,
302 now_epoch,
303 memory::STALE_MEMORY_ARCHIVE_DAYS,
304 )?,
305 archived_failures_to_purge: match policy.archived_failure_days {
306 Some(days) => db::count_archived_failures_to_purge_at(conn, now_epoch, days)?,
307 None => db::ArchivedFailurePurgePlan::default(),
308 },
309 })
310}
311
312fn apply_cleanup_plan(
313 conn: &Connection,
314 now_epoch: i64,
315 policy: CleanupPolicy,
316) -> Result<CleanupApplied> {
317 Ok(CleanupApplied {
318 expired_memories_marked_stale: memory::lifecycle::expire_active_memories(conn, now_epoch)?,
319 inactive_workstreams_paused: workstream::auto_pause_all_inactive_at(
320 conn,
321 now_epoch,
322 workstream::DEFAULT_AUTO_PAUSE_DAYS,
323 )?,
324 long_paused_workstreams_abandoned: workstream::auto_abandon_all_inactive_at(
325 conn,
326 now_epoch,
327 workstream::DEFAULT_AUTO_ABANDON_DAYS,
328 )?,
329 old_events_deleted: memory::cleanup_old_events_at(
330 conn,
331 now_epoch,
332 memory::OLD_EVENT_RETENTION_DAYS,
333 )?,
334 compressed_source_observations_deleted: memory::cleanup_compressed_source_observations_at(
335 conn,
336 now_epoch,
337 memory::COMPRESSED_SOURCE_OBSERVATION_RETENTION_DAYS,
338 )?,
339 stale_memories_archived: memory::archive_stale_memories_at(
340 conn,
341 now_epoch,
342 memory::STALE_MEMORY_ARCHIVE_DAYS,
343 )?,
344 archived_failures_purged: match policy.archived_failure_days {
345 Some(days) => db::purge_archived_failures_at(conn, now_epoch, days)?,
346 None => db::ArchivedFailurePurgePlan::default(),
347 },
348 })
349}
350
351fn validate_automatic_job(
352 conn: &Connection,
353 job_id: i64,
354 lease_owner: &str,
355 now_epoch: i64,
356) -> Result<()> {
357 let valid: bool = conn.query_row(
358 "SELECT EXISTS(
359 SELECT 1 FROM jobs
360 WHERE id = ?1 AND job_type = 'cleanup' AND state = 'processing'
361 AND lease_owner = ?2 AND lease_expires_epoch IS NOT NULL
362 AND lease_expires_epoch >= ?3
363 )",
364 params![job_id, lease_owner, now_epoch],
365 |row| row.get(0),
366 )?;
367 if !valid {
368 bail!("automatic cleanup job lease validation failed: job_id={job_id} owner={lease_owner}");
369 }
370 Ok(())
371}
372
373fn finish_automatic_job(
374 conn: &Connection,
375 job_id: i64,
376 lease_owner: &str,
377 lease_validation_epoch: i64,
378 finished_at_epoch: i64,
379) -> Result<()> {
380 let changed = conn.execute(
381 "UPDATE jobs
382 SET state = 'done', lease_owner = NULL, lease_expires_epoch = NULL,
383 next_retry_epoch = 0, last_error = NULL, failure_class = NULL,
384 failed_at_epoch = NULL, archived_at_epoch = NULL,
385 updated_at_epoch = ?1
386 WHERE id = ?2 AND job_type = 'cleanup' AND state = 'processing'
387 AND lease_owner = ?3 AND lease_expires_epoch IS NOT NULL
388 AND lease_expires_epoch >= ?4",
389 params![
390 finished_at_epoch,
391 job_id,
392 lease_owner,
393 lease_validation_epoch
394 ],
395 )?;
396 if changed != 1 {
397 bail!("automatic cleanup job completion lost its lease: job_id={job_id}");
398 }
399 Ok(())
400}
401
402fn insert_success_run(
403 conn: &Connection,
404 trigger: CleanupTrigger<'_>,
405 started_at_epoch: i64,
406 finished_at_epoch: i64,
407 applied: &CleanupApplied,
408) -> Result<()> {
409 let counts_json = serde_json::to_string(applied)?;
410 conn.execute(
411 "INSERT INTO maintenance_runs
412 (job_id, \"trigger\", policy_version, started_at_epoch,
413 finished_at_epoch, outcome, counts_json, error)
414 VALUES (?1, ?2, ?3, ?4, ?5, 'success', ?6, NULL)",
415 params![
416 trigger.job_id(),
417 trigger.as_str(),
418 CLEANUP_POLICY_VERSION,
419 started_at_epoch,
420 finished_at_epoch,
421 counts_json
422 ],
423 )
424 .context("record successful cleanup run")?;
425 Ok(())
426}
427
428fn record_failure_after_rollback(
429 conn: &Connection,
430 trigger: CleanupTrigger<'_>,
431 started_at_epoch: i64,
432 error: anyhow::Error,
433) -> anyhow::Error {
434 let finished_at_epoch = cleanup_finished_at(started_at_epoch);
435 let safe_error = safe_cleanup_error(&error);
436 let ledger_result = record_failure_run(
437 conn,
438 trigger,
439 started_at_epoch,
440 finished_at_epoch,
441 &safe_error,
442 );
443 match ledger_result {
444 Ok(()) => anyhow::anyhow!(safe_error),
445 Err(ledger_error) => {
446 let safe_ledger_error = safe_cleanup_error(&ledger_error);
447 anyhow::anyhow!(
448 "{safe_error}; additionally failed to record cleanup failure: {safe_ledger_error}"
449 )
450 }
451 }
452}
453
454fn record_failure_run(
455 conn: &Connection,
456 trigger: CleanupTrigger<'_>,
457 started_at_epoch: i64,
458 finished_at_epoch: i64,
459 safe_error: &str,
460) -> Result<()> {
461 let tx = Transaction::new_unchecked(conn, TransactionBehavior::Immediate)
462 .context("begin cleanup failure ledger transaction")?;
463 let job_id = match trigger.job_id() {
464 Some(job_id) if job_exists(&tx, job_id)? => Some(job_id),
465 _ => None,
466 };
467 let stored_error = if safe_error.is_empty() {
468 "cleanup transaction failed"
469 } else {
470 safe_error
471 };
472 tx.execute(
473 "INSERT INTO maintenance_runs
474 (job_id, \"trigger\", policy_version, started_at_epoch,
475 finished_at_epoch, outcome, counts_json, error)
476 VALUES (?1, ?2, ?3, ?4, ?5, 'failure', NULL, ?6)",
477 params![
478 job_id,
479 trigger.as_str(),
480 CLEANUP_POLICY_VERSION,
481 started_at_epoch,
482 finished_at_epoch,
483 stored_error
484 ],
485 )
486 .context("record failed cleanup run")?;
487 tx.commit()
488 .context("commit cleanup failure ledger transaction")?;
489 Ok(())
490}
491
492fn cleanup_finished_at(started_at_epoch: i64) -> i64 {
493 chrono::Utc::now().timestamp().max(started_at_epoch)
494}
495
496pub(crate) fn safe_cleanup_error(error: &anyhow::Error) -> String {
497 let raw = format!("{error:#}");
498 let bounded =
499 crate::adapter::common::redact_hook_payload_preview(&raw, FAILURE_ERROR_LIMIT_BYTES);
500 let bounded = bounded.trim();
501 if bounded.is_empty() {
502 "cleanup transaction failed".to_string()
503 } else {
504 bounded.to_string()
505 }
506}
507
508fn job_exists(conn: &Connection, job_id: i64) -> Result<bool> {
509 Ok(conn.query_row(
510 "SELECT EXISTS(SELECT 1 FROM jobs WHERE id = ?1)",
511 params![job_id],
512 |row| row.get(0),
513 )?)
514}
515
516#[cfg(test)]
517mod tests;